ShortIdImpl.C

Go to the documentation of this file.

// Copyright (C) 2001, Compaq Computer Corporation
// 
// This file is part of Vesta.
// 
// Vesta is free software; you can redistribute it and/or
// modify it under the terms of the GNU Lesser General Public
// License as published by the Free Software Foundation; either
// version 2.1 of the License, or (at your option) any later version.
// 
// Vesta is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
// Lesser General Public License for more details.
// 
// You should have received a copy of the GNU Lesser General Public
// License along with Vesta; if not, write to the Free Software
// Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

//
// ShortIdImpl.C
// Last modified on Mon Dec  5 17:49:09 EST 2005 by ken@xorian.net   
//      modified on Tue Aug  7 19:40:28 PDT 2001 by mann   
//      modified on Tue May  4 10:50:51 PDT 1999 by heydon 
//
// Server implementations of remote methods in ShortIdBlock.H
// 

#if __linux__
#include <stdint.h>
#endif
#include <sys/types.h>
#include <dirent.h>
// add declaration to fix broken <dirent.h> header file
extern "C" int _Preaddir_r(DIR *, struct dirent *, struct dirent **);

#include <pthread.h>
#include <time.h>
#include <stdlib.h>
#include <sys/stat.h>
#include <ctype.h>

#include <VestaConfig.H>
#include <Thread.H>

#include "ShortIdImpl.H"
#include "VestaLog.H"
#include "Recovery.H"
#include "ReadersWritersLock.H"
#include "ShortIdKey.H"
#include "FdCache.H"
#include "VRConcurrency.H"
#include "logging.H"

#include "lock_timing.H"

using std::fstream;
using std::hex;
using std::dec;

// Pitfall: on Tru64 random's state is per-thread, so it would be
// difficult to make sure it gets reseeded (with a distinct seed!)
// in every thread.  Instead we use the less-standard random_r function.
#define USE_RANDOM_R 1

// Types
struct ShortIdBlockInfo {
    time_t leaseExpires;
};
typedef Table<ShortIdKey, ShortIdBlockInfo*>::Default BlockInfoTable;
typedef Table<ShortIdKey, ShortIdBlockInfo*>::Iterator BlockInfoIter;

// Parameters
const int LEASE_PERIOD = 60*60*24;  // validity period for leases (sec)
const int LANDLORD_SLEEP = 60*60;   // period between lease checks (sec)
const int LANDLORD_WORKLIST_SIZE = 128;

// Module globals
static BlockInfoTable biTable;
#if USE_RANDOM_R
/*static*/ struct random_data randDat;
/*static*/ int randState[32];
#endif
static Basics::mutex mu;          // protects biTable, randDat
static Basics::thread landlord;   // detects expired leases
static Text sid_dir;

// RPC to get a new block of ShortIds from the server, all with the
// specified leafflag.  On return, the block is leased to the caller
// until bk.leaseExpires (set by the function).  If the lease expires
// without being renewed (below), the server will eventually notice
// and reclaim all unused ShortIds in the block.  If local is true,
// the caller is in the same address space as the server, so the lease
// is nonexpiring and is not logged---if the server crashes, the
// caller crashes too, and no longer needs the lease.
void
AcquireShortIdBlock(ShortIdBlock& bk, bool leafflag, bool local) throw()
{
    ShortId start;
    int used, used2 = -1;
    ShortIdBlock bk2;
    
    if (!local) {
        StableLock.acquireWrite(); // protects the log
        RWLOCK_LOCKED_REASON(&StableLock, "AcquireShortIdBlock");
    }
    mu.lock();
    for (;;) {
        // Randomly choose a nonzero start for the block
        int randint;
#if USE_RANDOM_R
# if __digital__
        (void) random_r(&randint, &randDat);
# else
        (void) random_r(&randDat, &randint);
# endif
#else
        randint = random();
#endif
        start = (ShortId)
          ( (randint & ~(ShortIdBlock::size - 1) &
             ~ShortIdBlock::leafFlag & ~ShortIdBlock::dirFlag)
           | (leafflag ? ShortIdBlock::leafFlag : 0) );
        if (start == NullShortId) {
            continue;
        }
        ShortIdBlockInfo* dummy;
        if (biTable.Get(start, dummy)) {
            // Some other process is using this block; try another
            continue;
        }

        // Check that there are some free ShortIds there
        bk.init(start);
        used = 0;
        char *name = ShortIdBlock::shortIdToName(start);
        char *p = strrchr(name, PathnameSep);
        assert(p != 0);
        *p = '\0';
        DIR *dir = opendir(name);
        delete[] name;
        if (!dir) {
            // Directory not present, so all of block is free
            break;
        }
        struct dirent de, *done;
        while (readdir_r(dir, /*OUT*/ &de, /*OUT*/ &done) == 0 &&
               done != (struct dirent *)NULL) {
            char *endptr;
            long offset = strtol(de.d_name, &endptr, 16);
            if (*endptr != '\0') continue;
            bk.set(start + offset);
            used++;
        }
        closedir(dir);
        if (used >= ShortIdBlock::size) {
            // Nothing free in this block; try another
            continue;
        }
        if (used2 == -1 && used > ShortIdBlock::size / 2) {
            // Block more than half full; save it and try another
            bk2 = bk;
            used2 = used;
            continue;
        } else {
            break;
        }
    }
    if (used2 != -1) {
        if (used2 < used) {
            // First one tried was less full!
            bk = bk2;
        }
    }
    ShortIdKey key(start);
    ShortIdBlockInfo* bi = NEW(ShortIdBlockInfo);
    bk.leaseExpires = bi->leaseExpires =
      (local ? ShortIdBlock::leaseNonexpiring : time(NULL) + LEASE_PERIOD);
    biTable.Put(key, bi);
    if (!local) {
        char logrec[256];
        // ('asidb' start leaseExpires)
        int sres = sprintf(logrec, "(asidb 0x%x %d)\n",
                           start, bk.leaseExpires);
        assert(sres > 0);
        assert(sres < sizeof(logrec));
        VRLog.start();
        VRLog.put(logrec);
        VRLog.commit();
    }
    mu.unlock();
    if (!local) {
        StableLock.releaseWrite();
    }
}


// RPC to renew the lease on a block.  Returns false if the block's
// lease had already expired.  (Allowing a lease to expire is a fatal
// client error.)  Otherwise, returns true, and the block is leased to
// the caller until block.leaseExpires (set by the function).  This
// routine is not called for local (nonexpiring) leases.
//
bool
RenewShortIdBlock(ShortIdBlock& bk) throw()
{
    StableLock.acquireWrite();
    RWLOCK_LOCKED_REASON(&StableLock, "RenewShortIdBlock");
    mu.lock();
    ShortIdKey key(bk.start);
    ShortIdBlockInfo* bi;
    bool inTable = biTable.Delete(key, bi); // reuse old bi value
    time_t now = time(0);
    if (!inTable || bi->leaseExpires < now) {
        mu.unlock();
        StableLock.releaseWrite();
        if(inTable) delete bi;
        return false;
    }
    bk.leaseExpires = bi->leaseExpires = now + LEASE_PERIOD;
    biTable.Put(key, bi);
    char logrec[256];
    // ('asidb' start leaseExpires)
    int sres = sprintf(logrec, "(asidb 0x%x %d)\n",
                       bk.start, bk.leaseExpires);
    assert(sres > 0);
    assert(sres < sizeof(logrec));
    VRLog.start();
    VRLog.put(logrec);
    VRLog.commit();
    mu.unlock();
    StableLock.releaseWrite();
    return true;
}


static void
ReleaseShortIdBlockAt(ShortId start, bool local) throw()
{
    ShortIdBlockInfo* bi;
    bool ok = biTable.Delete(start, bi);
    if (ok) delete bi;
    if (!local) {
        char logrec[256];
        // ('rsidb' start)
        int sres = sprintf(logrec, "(rsidb 0x%x)\n", start);
        assert(sres > 0);
        assert(sres < sizeof(logrec));
        VRLog.start();
        VRLog.put(logrec);
        VRLog.commit();
    }
}

// RPC to return a block to the server
// The caller promises not to assign any more ShortIds from this block, even
//  if some are still unused.  This routine should be called when a process
//  is going to shut down, or when it has used up a block. If local is true,
//  the caller is in the same address space as the server, so the lease
//  is not logged---if the server crashes, the caller crashes too, and
//  no longer needs the lease.
//
void
ReleaseShortIdBlock(ShortIdBlock& bk, bool local) throw()
{
    if (!local) {
        StableLock.acquireWrite();      // protects the log too
        RWLOCK_LOCKED_REASON(&StableLock, "ReleaseShortIdBlock");
    }
    mu.lock();
    ReleaseShortIdBlockAt(bk.start, local);
    mu.unlock();
    if (!local) {
        StableLock.releaseWrite();
    }
}


// 
// Thread to clean up blocks with expired leases
//
// Annoyingly, one can't delete from a Table while in the scope
// of an interator.  We would really like to delete the element
// the iterator has just returned sometimes.  Instead, we make an
// array of things to delete and do them later.  If the array fills,
// we get the rest on the next run.
//
// [Note: Tables have been since been improved to provide a way of
//  deleting while in the scope of an iterator, but the code below
//  has not been rewritten to take advantage of the new feature.]
//
void *
LandlordThread(void *arg)
{
    signal(SIGPIPE, SIG_IGN);
    signal(SIGQUIT, SIG_DFL);
    signal(SIGSEGV, SIG_DFL);
    signal(SIGABRT, SIG_DFL);
    signal(SIGILL, SIG_DFL);
    signal(SIGBUS, SIG_DFL);

    for (;;) {
        ShortId worklist[LANDLORD_WORKLIST_SIZE];
        int i = 0;
        StableLock.acquireWrite();
        RWLOCK_LOCKED_REASON(&StableLock, "LandlordThread");
        mu.lock();
        {
            BlockInfoIter iter(&biTable);
            ShortIdKey key;
            ShortIdBlockInfo* bi;
            time_t now = time(NULL);
            while (iter.Next(key, bi)) {
                if (bi->leaseExpires < now) {
                    worklist[i++] = key.sid;
                    if (i >= LANDLORD_WORKLIST_SIZE) break;
                }
            }
        }
        int j;
        for (j=0; j<i; j++) {
            ReleaseShortIdBlockAt(worklist[j], false);
        }
        mu.unlock();
        StableLock.releaseWrite();
        sleep(LANDLORD_SLEEP);
    }

    //return (void *)NULL;     // Not reached
}

// Code to rebuild biTable from the log
static void
AsidbCallback(RecoveryReader* rr, char &c)
     throw(VestaLog::Error, VestaLog::Eof)
{
    long ltmp;
    ShortId start;
    time_t leaseExpires;
    RecoveryReader::Ident id;
    
    // start leaseExpires
    rr->getLong(c, ltmp);
    start = (ShortId) ltmp;
    rr->getLong(c, ltmp);
    leaseExpires = (ShortId) ltmp;
    
    ShortIdKey key(start);
    
    ShortIdBlockInfo* bi;
    bool inTable = biTable.Delete(key, bi);
    if(!inTable)
      bi = NEW(ShortIdBlockInfo);
    else
      assert(bi != 0);
    bi->leaseExpires = leaseExpires;
    mu.lock();
    inTable = biTable.Put(key, bi);
    assert(!inTable);
    mu.unlock();
}

static void
RsidbCallback(RecoveryReader* rr, char &c)
     throw(VestaLog::Error, VestaLog::Eof)
{
    ShortId start;
    long lstart;
    
    // start
    rr->getLong(c, lstart);
    start = (ShortId) lstart;
    mu.lock();
    ShortIdBlockInfo* bi;
    bool ok = biTable.Delete(start, bi);
    if (ok) delete bi;
    mu.unlock();
}

// Code to do checkpointing
void
ShortIdBlockCheckpoint(fstream& ckpt) throw()
{
    // VRLock.write is already held
    mu.lock();
    BlockInfoIter iter(&biTable);
    ShortIdKey key;
    ShortIdBlockInfo* bi;

    while (iter.Next(key, bi)) {
        ckpt << "(asidb 0x" << hex << key.sid << dec
          << " " << bi->leaseExpires << ")\n";
    }

    mu.unlock();
}


static const int charsPerArc[] = { 3, 3, 2, 0 };
static const int MAX_CHARS_PER_ARC = 3;

extern "C"
{
  static int
  arccmp(const void* a, const void* b)
  {
    return strcmp((char*) a, (char*) b);
  }
}

static ShortId
NextSidToKeep(SourceOrDerived& sidstream)
{
    ShortId nextkeep;

    if (sidstream.eof()) {
        Repos::dprintf(DBG_SIDDEL, "at end of ShortIdsFile\n");
        nextkeep = NullShortId;
    } else {
        for (;;) {
            sidstream >> nextkeep;
            if (sidstream.fail()) {
                assert(sidstream.eof());
                Repos::dprintf(DBG_SIDDEL, "hit end of ShortIdsFile\n");
                nextkeep = NullShortId;
            } else if (nextkeep == NullShortId) {
                Repos::dprintf(DBG_ALWAYS, "error: NullShortId on keep list\n");
                continue; // skip the null
            }
            Repos::dprintf(DBG_SIDDEL, "next to keep: 0x%08x\n", nextkeep);
            break;
        }
    }
    return nextkeep;
}

// Scan a directory level in ShortId storage.  Return the number of
// entries remaining.  On error, return -(errno).  Delete any ShortIds
// not listed in the sidfile, and any subdirectories that are emptied.
static int
ScanDirForDelete(int levelnum, const char* dirname, Bit32 dirnum,
                 ShortId& nextkeep, SourceOrDerived& sidstream, time_t lease)
{
    typedef char SidArc[MAX_CHARS_PER_ARC + 1];
    SidArc *arcs = NEW_PTRFREE_ARRAY(SidArc, 1 << (MAX_CHARS_PER_ARC * 4));
    int narcs = 0;
    int i, nkept, ret;
    DIR* dir = opendir(dirname);
    if (dir == NULL) {
        Repos::dprintf(DBG_ALWAYS, "error opening sid directory, errno %d\n", errno);
        assert(errno != 0);
        return -errno;
    }
    struct dirent de, *done;
    while (readdir_r(dir, /*OUT*/ &de, /*OUT*/ &done) == 0 &&
           done != (struct dirent *)NULL) {
        bool ok = true;
        for (i = 0; i < charsPerArc[levelnum]; i++) {
            if (!isxdigit(de.d_name[i]) || isupper(de.d_name[i])) {
                ok = false;
                break;
            }
        }
        if (!ok || de.d_name[charsPerArc[levelnum]] != '\000') continue;
        strcpy(arcs[narcs++], de.d_name);
    }
    closedir(dir);
    if (narcs == 0) return 0;
    qsort(arcs, narcs, MAX_CHARS_PER_ARC + 1, arccmp);
    nkept = narcs;
    if (charsPerArc[levelnum + 1] == 0) {
        // At bottom level, can do file deletions
        for (i = 0; i < narcs; i++) {
            ShortId cursid = (dirnum << (charsPerArc[levelnum]*4)) +
              strtoul(arcs[i], NULL, 16);
            for (;;) {
                if (nextkeep != NullShortId && cursid > nextkeep) {
                    if (SourceOrDerived::dirShortId(nextkeep)) {
                        Repos::dprintf(DBG_SIDDISP, "directory: 0x%08x\n", nextkeep);
                    } else {
                        Repos::dprintf(DBG_ALWAYS, "missing: 0x%08x\n", nextkeep);
                    }
                    nextkeep = NextSidToKeep(sidstream);
                    continue;
                }
                break;
            }
            if (nextkeep == cursid) {
                Repos::dprintf(DBG_SIDDISP, "listed: 0x%08x\n", cursid);
                nextkeep = NextSidToKeep(sidstream);
            } else {
                // cursid < nextkeep || nextkeep == NullShortId
                char *sidname = NEW_PTRFREE_ARRAY(char, (strlen(dirname) + 1 +
                                                         strlen(arcs[i]) + 1));
                if (*dirname == '\0') {
                    strcpy(sidname, arcs[i]);
                } else {
                    sprintf(sidname, "%s%c%s",
                            dirname, PathnameSep, arcs[i]);
                }
                struct stat statbuf;
                int res = stat(sidname, &statbuf);
                if (res < 0) {
                    if (errno == ENOENT) {
                      // Harmless race with eager deletion of this sid
                      nkept--;
                      delete [] sidname;
                      continue;
                    }
                    Repos::dprintf(DBG_ALWAYS,
                                   "error on stat of sid, errno %d\n", errno);
                    assert(errno != 0);
                    delete [] sidname;
                    return -errno;
                }
                if (statbuf.st_ctime < lease) {
                    Repos::dprintf(DBG_SIDDISP, "garbage: 0x%08x\n", cursid);
                    if (!Repos::isDebugLevel(DBG_SIDNODEL)) {
                        res = unlink(sidname);
                        if (res < 0 && errno != ENOENT) {
                            Repos::dprintf(DBG_ALWAYS,
                                           "error on unlink of sid, errno %d\n", errno);
                            assert(errno != 0);
                            delete [] sidname;
                            return -errno;
                        }
                    }
                    FdCache::flush(cursid, FdCache::any);
                    nkept--;
                } else {
                    Repos::dprintf(DBG_SIDDISP, "leased: 0x%08x\n", cursid);
                }
                delete [] sidname;
            }
        }
    } else {
        for (i = 0; i < narcs; i++) {
            char *subdirname = NEW_PTRFREE_ARRAY(char, (strlen(dirname) + 1 +
                                                        strlen(arcs[i]) + 1));
            Bit32 subdirnum = (dirnum << (charsPerArc[levelnum]*4)) +
              strtoul(arcs[i], NULL, 16);
            if (*dirname == '\0') {
                strcpy(subdirname, arcs[i]);
            } else {
                sprintf(subdirname, "%s%c%s",
                        dirname, PathnameSep, arcs[i]);
            }
            ret = ScanDirForDelete(levelnum + 1, subdirname, subdirnum,
                                   nextkeep, sidstream, lease);
            if (ret < 0) {
              delete [] subdirname;
              return ret;
            }
            if (ret == 0) {
                Repos::dprintf(DBG_SIDDISP, "empty: %s\n", subdirname);
                if (!Repos::isDebugLevel(DBG_SIDNODEL)) {
                    int res = rmdir(subdirname);
                    if (res < 0) {
                        Repos::dprintf(DBG_ALWAYS, "error on sid rmdir, errno %d\n",
                                       errno);
                        assert(errno != 0);
                        delete [] subdirname;
                        return -errno;
                    }
                }
                nkept--;
            }
            delete [] subdirname;
        }
    }
    delete [] arcs;
    return nkept;
}


// Do actual deletions, at the end of a weed.
// sidfile must be sorted.
int
DeleteAllShortIdsBut(ShortIdsFile sidfile, time_t lease) throw()
{
    ShortId nextkeep;
    SourceOrDerived sidstream;
    sidstream.open(sidfile);
    sidstream >> hex;
    if (!sidstream.good()) {
        int ret = errno;
        Repos::dprintf(DBG_ALWAYS,
                       "error opening combined keepSidFile, errno %d\n", ret);
        assert(ret != 0);
        sidstream.close();
        return ret;
    }
    // Prime the pump
    nextkeep = NextSidToKeep(sidstream);
    int ret =
      ScanDirForDelete(0, sid_dir.cchars(), 0, nextkeep, sidstream, lease); 
    sidstream.close();
    if (ret < 0) return -ret;
    return 0;
}


void
ShortIdServerInit()
{
#if USE_RANDOM_R
    memset((void *) randState, 0, sizeof(randState));
    randDat.state = NULL;
    char *dummy;
# if __digital__
    (void) initstate_r((unsigned) time(0),
                       (char *) randState, sizeof(randState),
                       &dummy, &randDat);
# else
    (void) initstate_r((unsigned) time(0),
                       (char *) randState, sizeof(randState),
                       &randDat);
# endif
#else
    srandom((unsigned) time(NULL));
#endif //NOTDEF
    RegisterRecoveryCallback("asidb", AsidbCallback);
    RegisterRecoveryCallback("rsidb", RsidbCallback);
    try {
        bool ok = VestaConfig::get("Repository", "sid_dir", sid_dir);
        if (!ok) sid_dir = "";
    } catch (VestaConfig::failure f) {
        Repos::dprintf(DBG_ALWAYS, "VestaConfig::failure %s\n", f.msg.cchars());
        abort();
    }
}

void
ShortIdServerInit2()
{
  Basics::thread_attr landlord_attr;
#if defined (_POSIX_THREAD_PRIORITY_SCHEDULING) && !defined(__linux__)
  // Linux only allows the superuser to use SCHED_RR
  landlord_attr.set_schedpolicy(SCHED_RR);
  landlord_attr.set_inheritsched(PTHREAD_EXPLICIT_SCHED);
  landlord_attr.set_sched_priority(sched_get_priority_min(SCHED_RR));
#endif
  
  landlord.fork(LandlordThread, 0, landlord_attr);
}

---