Bdb packages | Design docs | Source docs | Guidelines | Recent releases
	Search | Site Map .

                 

---

/BdbDistTools/BdbDistribution.cc

Go to the documentation of this file.

//--------------------------------------------------------------------------
// File and Version Information:
//      $Id: BdbDistribution.cc,v 1.20 2002/06/17 03:00:04 becla Exp $
//
// Description:
//      Support for the distribution operations (attach, extract, ...)
//
// Environment:
//      Software developed for the BaBar Detector at the SLAC B-Factory.
//
// Author List:
//      Jean-Noel Albert                        Original Author
//
// Copyright Information:
//      Copyright (C) 1999
//
//------------------------------------------------------------------------

//-----------------------
// This Class's Header --
//-----------------------
#include "BdbDistTools/BdbDistribution.hh"

//-------------
// C Headers --
//-------------
extern "C" {
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <strings.h>
#include <sys/wait.h>
}

//-------------------------------
// Collaborating Class Headers --
//-------------------------------
#include "BdbApplication/BdbDomain.hh"
#include "BdbEvent/BdbEvent.hh"
#include "BdbEvent/BdbEventT.hh"
#include "BdbEventStore/BdbEventStore.hh"
#include "BdbEventStore/BdbAbsCollectionT.hh"
#include "BdbEventStore/BdbCollectionP.hh"
#include "BdbTrees/BdbTreeAccess.hh"
#include "BdbAccess/BdbTFileSystem.hh"
#include "BdbAccess/BdbDbInfo.hh"
#include "BdbTrees/BdbDbRegistrator.hh"
#include "BdbAccess/BdbAuth.hh"
#include "BdbAccess/BdbDbTreeSingleton.hh"
#include "BdbDistTools/BdbDistMessage.hh"
#include "BdbEvent/BdbEvent_001.hh"

BdbDistribution::BdbDistribution()
{
//    setVerbose(BdbApplication::instance()->verbose());
}

BdbDistribution::~BdbDistribution()
{}

//------------
// Services --
//------------
void BdbDistribution::setVerbose(d_Boolean mode)
{
//    BdbApplication::instance()->setVerbose(mode);
}

d_Boolean BdbDistribution::verbose () const
{
//    return BdbApplication::instance()->verbose();
   return( false ) ;
}

BdbTString BdbDistribution::logfile(const char* name, const char* suffix)
{
    char* tmp = getenv("BDBDIST_TMP");
    if (0 == tmp)
        tmp = "/tmp";
    BdbTString logname(tmp, "/", name, "-", BdbTString(getpid()), suffix);

    // Remove any existing file
    if (access(logname, F_OK) == 0)
        unlink(logname);

    return logname;
}

// Translate BdbDistribution verbose, message, and error calls to
// BdbDistMessage.
// Note that BdbDistribution::message() indicates an error condition.

void BdbDistribution::verbose(BdbTString text)
{
  BdbDistMessage::verbose (text.data());
}

void BdbDistribution::message(BdbTString text)
{
  BdbDistMessage::warning (text.data());
}

BdbStatus BdbDistribution::error(BdbTString text)
{
    // Print the error message
    message(text);

    // Cancel the *** internal *** transaction, if any
    if (!_externalTransaction && !checkTransaction(BdbcNoOpen))
        BdbDomain::activeInstance()->abort("BdbDistribution::error");

    // Signal the error
    return BdbcError;
}

int BdbDistribution::execute(BdbTString command)
{
    verbose ("> " + command);
    int stt = system(command);
    return WEXITSTATUS(stt);
}

int BdbDistribution::catalogOperation(BdbTString command)
{
    // Drop any lock before executing the operation
    int mode = getTransaction();
    if (BdbcNoOpen != mode)
        BdbDomain::activeInstance()->commit("BdbDistribution::catalogOperation");

    // Execute the catalog operation
    int stt = execute(command);

    // Reset the transaction in the initial mode
   if (BdbcRead == mode)
        BdbDomain::activeInstance()->startRead("BdbDistribution::catalogOperation");
    else if (BdbcUpdate == mode)
        BdbDomain::activeInstance()->startUpdate("BdbDistribution::catalogOperation");
   return stt;
}

d_Boolean BdbDistribution::checkTransaction(int mode)
{
    _externalTransaction = BdbcNoOpen != BdbDomain::activeInstance()->mode();
    if (!_externalTransaction)
        return BdbcNoOpen == mode;      // No transaction at all
                                        // Return true if asked for "No Open"
    else if (BdbcNoOpen == mode)
        return d_False;                 // Inside of a transaction: wrong

    else if (BdbcUpdate == mode)
        return (mode == BdbDomain::activeInstance()->mode());

    return d_True;                      // At least in read mode
}

int BdbDistribution::getTransaction()
{
    return BdbDomain::activeInstance()->mode();
}

//--------------
// Operations --
//--------------

/*********************************************************************************/

// Context: NO TRANSACTION
//              *** existing transaction is aborted ***
//
// Access: Any
//
// Privileges: database authorization level

BdbStatus BdbDistribution::attachDb (BdbDistFile& filepath, int dbid,
                                     d_Boolean chmodRead, 
                                     BdbTString dbname, d_Boolean force)
{
    verbose("Attaching " + filepath.name() + " ...");

    // Check for the transaction mode
    if (!checkTransaction(BdbcNoOpen))
        return error("Can't work inside a transaction");

    // Start an local transaction to be able to check for the privileges
    BdbDomain::activeInstance()->startRead("BdbDistribution::attachDb");    

    // Preliminary checks
    if (!filepath.isFileExist())
        return error("No such file: " + filepath.fullPath());

    if (!force && "" != dbname && filepath.name() != dbname)
        return error ("Not allowed to attach " + filepath.fullPath() +
                      " as: " + dbname);

    if ("" == dbname)
        dbname = filepath.name();

    BdbDistDatabase db(dbname.data());
    if (db.isDbExist())
        return error ("Database " + db.dbName() + 
                      " already exist: " + BdbTString(db.dbid()));

    if (!filepath.isAuthorized())
        return error ("No privilege to attach: " + filepath.fullPath());

    // Name an error file to receive the output of "ooattachdb"
    BdbTString errlog = logfile(dbname, ".err");

    // Attach command
    BdbTString command = "ooattachdb";
    if (! verbose())
        command += " -notitle -quiet";
    command += " -db " + db.dbName();
    command += " -id " + BdbTString(dbid);
    command += " -filepath " + filepath.filePath();
    if ("" != filepath.host())
        command += " -host " + filepath.host();

    command += " ";
    command += BdbDomain::activeInstance()->bootName();

    if (!verbose())
        command += " 2> " + errlog;

    // Execute the ooattachdb command
    if (0 != catalogOperation(command)) {
        message ("Attach command failed: " + command);
        if (!verbose())
            execute("cat " + errlog);
        return BdbcError;
    }

    // Remove the error log file
    unlink(errlog);

    // Turn the file read/only
    if (chmodRead && BdbcSuccess != filepath.chmodRead())
        return error("Failed to read protect: " + filepath.fullPath());

    // Commit the local transaction
    BdbDomain::activeInstance()->commit("BdbDistribution::attachDb");

    return BdbcSuccess;
}

/*********************************************************************************/
/*********************************************************************************/

// Context: NO TRANSACTION
//              *** existing transaction is aborted ***
//
// Access: Any
//
// Privileges: database authorization level

BdbStatus BdbDistribution::detachDb (BdbTString dbname, int retry, int delay)
{
    verbose("Detaching " + dbname + " ...");

    // Check for the transaction mode
    if (!checkTransaction(BdbcNoOpen))
        return error("Can't work inside a transaction");

    // Start an local transaction to be able to check for the privileges
    BdbDomain::activeInstance()->startRead("BdbDistribution::detachDb");

    // Privileges
    BdbDistDatabase db (dbname.data());
    if (!db.isAuthorized())
        return error("No privilege to detach " + dbname);

    // Error log file to keep the oodetachdb message
    BdbTString errlog = logfile(dbname, ".err");

    // Detach command
    BdbTString command = "oodeletedb -force -catalogonly";
    if (!verbose())
        command += " -notitle -quiet";
    command += " -db " + dbname;
    command += " ";
    command += BdbDomain::activeInstance()->bootName();

    if (!verbose())
        command += " 2> " + errlog;

    // Execute the detach command
    if (0 != catalogOperation(command)) {
        message ("Detach command failed: " + command);
        if (!verbose())
            execute ("cat " + errlog);

        for (int t = 0; t < retry; t++) {
            message ("Retrying in " + BdbTString(delay) + " seconds...");
            sleep(delay);
            if (0 == catalogOperation(command)) {
                verbose("Success !");
                break;
            }

            message ("Detach command failed: " + command);
            if (!verbose())
                execute ("cat " + errlog);
        }

        if (retry >= 0)
            message ("Retry limit exceeded...");

        return BdbcError;
    }

    // Remove the error log file
    unlink(errlog);

    // Commit the local transaction
    BdbDomain::activeInstance()->commit("BdbDistribution::detachDb");

    return BdbcSuccess;
}

/*********************************************************************************/
/*********************************************************************************/

// Context: Update Transaction
//
// Access: Update
//
// Privileges: database authorization level

BdbStatus BdbDistribution::changeDb (BdbDistDatabase& db,
                                     BdbDistFile& filepath,
                                     d_Boolean chmodRead, d_Boolean force)
{
    verbose ("Changing " + db.dbName() + " ...");

    // Check for the transaction mode
    if (!checkTransaction(BdbcUpdate))
        return error ("Need an update transaction");

    // Preliminary checks
    if (!db.isDbExist())
        return error ("No such database: " + db.dbName());

    // Existing file
    if (!force && !filepath.isFileExist())
        return error ("No such file: " + filepath.fullPath());

    // Check the system names
    if (!force && db.dbName() != filepath.name())
        return error ("Not allowed to change: " + db.dbName() +
                      " as: " + filepath.fullPath());

    // Privileges
    if (!db.isAuthorized())
        return error ("No privilege to change: " + db.dbName());

    // Lock the database in update mode to be able to change it
    //  The caller has to define the wait delay...
    verbose ("Trying to lock: " + db.dbName() + " ...");
    if (BdbcSuccess != db.lock(oocLockUpdate))
        return error ("Can't lock for update: " + db.dbName());

    verbose ("Lock granted...");

    // Send the Objectivity outputs in an error log file
    BdbTString errlog = logfile(db.dbName(), ".err");
    FILE* flog = NULL;

    if (!verbose())
        flog = fopen(errlog, "w");
    if (NULL == flog)
        flog = stdout;

    // Execute the change operation
    BdbTString host;
    if ("" == filepath.host())
        host = BdbTFileSystem::localHostName();
    else
        host = filepath.host();

//     BdbStatus status = db.handle().change(db.dbName(), host, filepath.filePath(), flog);
    BdbRef(BdbDBObj) dbR = db.handle();
    BdbStatus status = dbR.change(0, host, filepath.filePath(), flog);

    if (stdout != flog)
        fclose (flog);

    // Remove the error log, if the change success
    if (BdbcSuccess == status)
        unlink(errlog);

    // Or print it on the user screen
    else if (!verbose())
        execute ("cat " + errlog);

    if (BdbcSuccess != status)
        return status;

    // Turn the file read/only
    if (chmodRead && BdbcSuccess != filepath.chmodRead())
        return error("Failed to read protect: " + filepath.fullPath());

    return BdbcSuccess;
}

/*********************************************************************************/
/*********************************************************************************/

// Context: Read transaction
//
// Access: Any
//
// Privileges: none

BdbStatus BdbDistribution::extractDb (BdbDistDatabase& db,
                                      BdbDistFile& outfile,
                                      d_Boolean supersed)
{
    verbose ("Extracting " + db.dbName() + " ...");

    // Check for the transaction mode
    //  Start a local transaction if needed
    if (checkTransaction(BdbcNoOpen))
        BdbDomain::activeInstance()->startRead("BdbDistribution::extractDb");

    // Preliminary checks
    if (!db.isDbExist())
        return error ("No such database: " + db.dbName());

    if (!db.isFileExist())
        return error ("Database file does not exist: " + db.fullPath());

    // Existing destination file
    if (outfile.isFileExist()) {
        if (!supersed)
            return error ("Output file exists: " + outfile.fullPath());

        // Remove the output file
        if (BdbcSuccess != outfile.remove())
            return error ("Failed to remove the existing output file: " +
                          outfile.fullPath());
    }

    //  Lock the database in read mode to be able to change it
    //  The caller has to define the wait delay...
    verbose ("Trying to lock: " + db.dbName() + " ...");
    if (BdbcSuccess != db.lock(oocLockRead))
        return error ("Can't lock for read: " + db.dbName());

    verbose ("Lock granted...");

    // Get the size of the database -- to check the result of the transfer
    off_t size;
    if (BdbcSuccess != db.fileSize(size))
        return error ("Can't get the size of the database file: " + db.fullPath());

    // Copy the file -- no cross node at this time
    if (BdbcSuccess != db.copy(outfile))
        return error ("Failed to copy the database file: " + db.fullPath() +
                      " to " + outfile.fullPath());

    // Check if the out file exists, with the correct size
    if (!outfile.isFileExist())
        return error ("Ouput file does not exists: " + outfile.fullPath());

    off_t outsize;
    if (BdbcSuccess != outfile.fileSize(outsize))
        return error ("Failed to get the size of the output file: " +
                      outfile.fullPath());

    if (size != outsize)
        return error ("The output file size does not match the database size: " +
                       outfile.fullPath());

    // Finish the local transaction
    if (!_externalTransaction)
        BdbDomain::activeInstance()->commit("BdbDistribution::extractDb");
    return BdbcSuccess;
}

/*********************************************************************************/
/*********************************************************************************/

// Context: NO TRANSACTION
//
// Access: Update
//
// Privileges: database authorization level

BdbStatus BdbDistribution::removeDb (BdbTString dbname, int retry, int delay)
{
    verbose ("Removing " + dbname + " ...");

    // Check for the transaction mode
    if (!checkTransaction(BdbcUpdate))
        return error ("Need an update transaction");

    // Start a local transaction to be able to check for the privileges
    BdbDomain::activeInstance()->startRead("BdbDistribution::removeDb");

    // Preliminary checks
    BdbDistDatabase db (dbname.data());
    if (!db.isDbExist())
        return error ("No such database: " + dbname);

    // Privileges
    if (!db.isAuthorized())
        return error ("No privilege to remove: " + dbname);

    //  Lock the database in update mode
    //  The caller has to define the wait delay...
    verbose ("Trying to lock: " + dbname + " ...");
    if (BdbcSuccess != db.lock(oocLockUpdate))
        return error ("Can't lock for update: " + dbname);

    verbose ("Lock granted...");

    // Change the database to an invalid location, to prevent accesses
    BdbDistFile invfile ("/invalid/not/a/db");
    if (BdbcSuccess != changeDb(db, invfile, d_True)) {
        BdbDomain::activeInstance()->abort("BdbDistribution::detachDb");
        return error ("Remove failed");
    }

    // Commit the local transaction and try to detach the database
    BdbDomain::activeInstance()->commit("BdbDistribution::detachDb");

    return detachDb(dbname, retry, delay);
}

/*********************************************************************************/
/*********************************************************************************/

// Context: NO TRANSACTION
//              *** existing transaction is aborted ***
//
// Access: Update
//
// Privileges: database authorization level

BdbStatus BdbDistribution::shadowDb (BdbDistFile& filepath, int dbid)
{
    verbose ("Shadowing " + filepath.name() + " ...");

    // Check for the transaction mode
    if (!checkTransaction(BdbcNoOpen))
        return error("Can't work inside a transaction");

    // Start a local transaction
    BdbDomain::activeInstance()->startUpdate("BdbDistribution::shadowDb");

    // Preliminary checks
    if (filepath.isFileExist())
        return error ("Target file exists -- use \"attachDb\": " +
                      filepath.fullPath());

    BdbDistDatabase db(filepath.name().data());
    if (db.isDbExist())
        return error ("Database " + db.dbName() + 
                      " already exist: " + BdbTString(db.dbid()));

    if (!filepath.isAuthorized())
        return error ("No privilege to shadow: " + filepath.fullPath());

    // Create a dummy database in the federation directory
    verbose("Creating a dummy ." + filepath.name() + " database ...");
    BdbDomain* tempApp = BdbDomain::activeInstance();
    BdbDistFile dummyfile (BdbTString( tempApp->dirName(),
                                       "/.", db.dbName(), ".dummy"),
                           tempApp->hostName());
    if (dummyfile.isFileExist() && BdbcSuccess != dummyfile.remove())
        return error("Failed to remove existing dummy database: " +
                     dummyfile.fullPath());

    // Error log file to keep the oonewdb message
    BdbTString errlog = logfile(db.dbName(), ".err");

    // Create an empty database
    BdbTString command = "oonewdb -db " + db.dbName() +
                         " -host " + dummyfile.host() +
                         " -filepath " + dummyfile.filePath();
    if (!verbose())
        command += " -notitle -quiet";
    command += " ";
    command += BdbDomain::activeInstance()->bootName();

    if (!verbose())
        command += " 2> " + errlog;

    if (0 != execute(command))
        return error ("Failed to create the dummy database: " +
                      dummyfile.fullPath());

    // Detach the dummy and attach it with the right dbid
    //  Must be done out of a transaction
    BdbDistDatabase dummydb (db.dbName().data());

    BdbDomain::activeInstance()->commit("BdbDistribution::shadowDb");

    if (BdbcSuccess != detachDb(dummydb.dbName()))
        return error ("Can't detach the dummy database: " + db.dbName());

    // Attach with the right dbid
    if (BdbcSuccess != attachDb(dummyfile, dbid, d_False, db.dbName(), d_True))
        return error ("Can't reattach the dummy database: " + 
                      dummyfile.fullPath());

    // Restart a transaction to be able to change the database
    BdbDomain::activeInstance()->startUpdate("BdbDistribution::shadowDb");

    // Change the dummy to the new location
    if (BdbcSuccess != changeDb(dummydb, filepath, d_False, d_True)) {
        BdbDomain::activeInstance()->abort("BdbDistribution::shadowDb");
        return error ("Can't change the dummy: " + dummydb.dbName() +
                      " to the shadow location: " + filepath.fullPath());
    }

    // Remove the dummy
    if (BdbcSuccess != dummyfile.remove())
        message ("Failed to remove the dummy file: " + dummyfile.fullPath());

    // Final commit
    BdbDomain::activeInstance()->commit("BdbDistribution::shadowDb");

    return BdbcSuccess;
}

/*********************************************************************************/
/*********************************************************************************/

// Context: Update Transaction
//
// Access: Update
//
// Privileges: collection authorization level

BdbStatus BdbDistribution::replaceDb (BdbDistDatabase& db, BdbDistFile& impfile,
                                      d_Boolean chmodRead)
{
    verbose("Replacing  " + db.dbName() + " ...");

    // Check for the transaction mode
    if (!checkTransaction(BdbcUpdate))
        return error ("Need an update transaction");

    // Preliminary checks
    if (!db.isDbExist())
        return error ("No such database: " + db.dbName());

    // Existing file
    if (!impfile.isFileExist())
        return error ("No such file: " + impfile.fullPath());

    if (!db.isAuthorized())
        return error ("No privilege to attach: " + db.fullPath());

    // Lock the database in update mode to be able to change it
    //  The caller has to define the wait delay...
    verbose ("Trying to lock: " + db.dbName() + " ...");
    if (BdbcSuccess != db.lock(oocLockUpdate))
        return error ("Can't lock for update: " + db.dbName());

    verbose ("Lock granted...");

    // Rename the database file (if any)
    BdbDistFile olddb (db.filePath() + "old", db.host(), db.pudPort());

    if (db.isFileExist()) {
        verbose ("Renaming " + db.fullPath() + " as old");

//      BdbDistFile olddb (db.filePath() + "old");
        if (BdbcSuccess != db.rename(olddb))
            return error ("Failed to rename " + db.dbName() + 
                          " as " + olddb.filePath());
    }

    // Install the new database
    // Warning: the new file and the old one can be in a different path
//     BdbDistFile newfile (db.filePath());
    BdbDistFile newfile (impfile.directory() + "/" + db.fileName(), impfile.host(), impfile.pudPort());
    if (BdbcSuccess != impfile.rename(newfile))
        return error ("Failed to rename " + impfile.fullPath() +
                      " as " + newfile.fullPath());

    // Change the database to this new file
    changeDb(db, newfile, chmodRead);

    // Remove the old file
    if (olddb.isFileExist() && BdbcSuccess != olddb.remove())
        message ("Failed to remove the old " + olddb.fullPath());

    return BdbcSuccess;
}

/*********************************************************************************/
/*********************************************************************************/

// Context: Update Transaction
//
// Access: Update
//
// Privileges: collection authorization level

BdbStatus BdbDistribution::attachCollection (BdbTString pathname, BdbTString oid,
                                             d_Boolean force, 
                                             d_Boolean fixTreeNode)
{
    verbose("Attaching collection " + pathname + 
            " force " + BdbTString(force) + 
            " fixtreenode " + BdbTString(fixTreeNode));

    // Check for the transaction mode
    if (!checkTransaction(BdbcUpdate))
        return error ("Need an update transaction");

    // Convert the OID string to 4 numbers
    char* buffer = new char[oid.length() + 1];
    strcpy(buffer, oid);

    for (char* sep = buffer; *sep != '\0'; sep++)
        if (*sep == '-' || *sep == '#')
            *sep = ' ';

    int dbid, contid, pageid, slotid;
    if (sscanf(buffer, "%d %d %d %d", &dbid, &contid, &pageid, &slotid) != 4)
        return error ("Invalid collection id: " + oid);

    // Check if the collection exists
    //  If yes, and if the collection has the same "oid", it's OK
    //  else, if the "force" flag is not set, this is an error
    BdbHandle(BdbCollectionP) collH;
    BdbRef(BdbCollectionP) collR;

    BdbEventStore* estore = BdbEventStore::instance();
    d_Boolean deleteCollNode = false;

    BdbAbsCollectionT<BdbEventT>* collection = 
        estore->collection(pathname, (*(BdbEventT*)0));
    if (0 != collection) {
        collH = collection->persistent();
        collR = collH;
        
        // Check the collection OID
        if (collR.get_DB() == dbid && collR.get_OC() == contid &&
            collR.get_page() == pageid && collR.get_slot() == slotid) {
            // Same id. nothing to do unless user want to fix treenode
          if (! fixTreeNode ) {
            message("Collection " + pathname + 
                    " already exists with the same OID");
            return BdbcSuccess;
          }
        }
        else if (!force) {      
          // Different id; "force" not specified;
          //    It's an error
          return error ("Collection " + pathname + 
                        " already exist with a diff. OID");
        }
        else {
          // Different id; "force" specified;
          // Delete the collection
          //verbose("Removing obsolete collection " + pathname + " (diff. OID)");
          //if (BdbcSuccess != estore->removeCollection(collection))
          //    return error ("Failed to remove existing collection " + pathname);

          // AT: deletion of collection doesn't work if a collection was 
          // created in a different federation. Instead, we delete a collection
          // node from the treenodes (later, just before creation of a new collnode)
          
          deleteCollNode = true;
        }
    }

    // Check for the tree node root, or create it
    char* treepath = new char[pathname.length() + 1];
    strcpy(treepath, pathname);

    char* collnode = strrchr(treepath, '/');
    if (collnode == NULL)
        return error ("Invalid collection path: " + pathname);
    *collnode++ = '\0';

    BdbTreeNode node;
    BdbTreeAccess< BdbCollectionP > accessor;

    // Create a dummy reference for attaching the collection to this tree node
    collR.set_DB(dbid);         
    collR.set_OC(contid);
    collR.set_page(pageid);
    collR.set_slot(slotid);
    collH = collR;

    if (BdbcSuccess != node.moveTo(treepath)) {
        verbose("Creating tree node path " + BdbTString(treepath));

        // Set up the autorization level
//      if (BdbcSuccess != estore->BdbDomain::setAuthLevel((const char* const)treepath)) {
        if (BdbcSuccess != estore->BdbDomain::setAuthLevel(treepath)) {
            delete[] treepath;
            return error ("No privilege");
        }

        if (BdbcSuccess != node.makeTo(treepath)) {
            delete[] treepath;
            return error ("Failed to create the tree node " + BdbTString(treepath));
        }

    }
    else {
      if (deleteCollNode) {
        verbose("Deleting collnode " + BdbTString(collnode) + 
                " from " + BdbTString(treepath));
        if (BdbcSuccess != accessor.remove(node, collnode)) {
          return ("Failed to delete " + BdbTString(collnode) + 
                  " from " + BdbTString(treepath));
        }
      }
    }
    
    //attach child node 
    verbose("Attaching collection " + BdbTString(collnode));
    if (BdbcSuccess != accessor.add(node, collH, collnode)) {
      delete[] treepath;
      return ("Failed to insert the collection " + BdbTString(collnode));
    }
    
    // But unless we are in the original production federation
    // We don't want to do so, cause in the analysis federation 
    // this data must be readonly!! So, this operation is optional
    // Use BdbDistCollLoader -fixtreenode
    
    if (fixTreeNode) {
      verbose("Fixing treenode for " + BdbTString(treepath));
      BdbHandle(BdbTreeNodeP) nodeP = node.persistent();
      collH->setTreeNode(nodeP);
    }
    
    delete[] treepath;
    return BdbcSuccess;
}

/*********************************************************************************/
/*********************************************************************************/

// Context: Any
//              (the operation is done in an autonomous transaction)
//
// Access: n.a.
//
// Privileges: none

BdbStatus BdbDistribution::getFullDbs (ooTVArray(d_ULong)& dbs)
{
    verbose("Reading the full databases list ...");

    BdbEventStore& estore( *BdbEventStore::instance( ) );
    BdbEvsClusteringHint* hint = new BdbEvsClusteringHint(estore);
    return hint->getAllFullDbs(dbs);
}

/*********************************************************************************/
/*********************************************************************************/

// Context: Any
//
// Access: Any
//
// Privileges: System adminitrator

BdbStatus BdbDistribution::flushFullDbs (ooTVArray(d_ULong)& dbs)
{
    verbose("Flushingding the full databases list ...");

    // Check for the transaction mode
    if (checkTransaction(BdbcNoOpen))   
        BdbDomain::activeInstance()->startRead("BdbDistribution::flushFullDbs");

    // Check the user privileges
    BdbAuth* auth = BdbAuth::instance( );
    if (!auth->isAuthorized(BdbDomain::Events, BdbDomain::System))
        return error ("No privileges");

    BdbEventStore& estore( *BdbEventStore::instance( ) );
    BdbEvsClusteringHint* hint = new BdbEvsClusteringHint(estore);
    BdbStatus status = hint->removeFullDbsFromRegistry(dbs);

    // Commit the local transaction, if any
    if (!_externalTransaction)
        BdbDomain::activeInstance()->commit("BdbDistribution::flushFullDbs");

    return status;
}

/*********************************************************************************/
/*********************************************************************************/

// Context: Update Transaction
//
// Access: Update
//
// Privileges: database authorization level

BdbStatus BdbDistribution::clusteringRecordDb (BdbDistDatabase& db)
{
    verbose ("Registering " + db.dbName() + " ...");

    // Check for the transaction mode
    if (!checkTransaction(BdbcUpdate))
        return error ("Need an update transaction");

    // Preliminary checks
    if (!db.isDbExist())
        return error ("No such database: " + db.dbName());

    // Privileges
    if (!db.isAuthorized())
        return error ("No privilege to manage: " + db.dbName());

    BdbDbTreeSingleton* treesingleton = BdbDbTreeSingleton::instance();
    BdbDbInfo dbinfo (db.dbName(), db.host(), db.filePath());
    if (BdbcSuccess != treesingleton->registerDb(dbinfo))
        return error ("Failed to register " + db.dbName());

    return BdbcSuccess;
}

/*********************************************************************************/
/*********************************************************************************/

// This entry point is for the shadowed database
BdbStatus BdbDistribution::clusteringRecordDb (const char* dbname, const char* host, 
                                               const char* filepath)
{
    verbose ("Registering " + BdbTString(dbname) + " ...");

    // Check for the transaction mode
    if (!checkTransaction(BdbcUpdate))
        return error ("Need an update transaction");

    // Privileges
    BdbDistFile file(filepath);
    if (!file.isAuthorized())
        return error ("No privilege to manage: " + BdbTString(dbname));

    BdbDbTreeSingleton* treesingleton = BdbDbTreeSingleton::instance();
    BdbDbInfo dbinfo (dbname, host, filepath);
    if (BdbcSuccess != treesingleton->registerDb(dbinfo))
        return error ("Failed to register " + BdbTString(dbname));

    return BdbcSuccess;
}

/*********************************************************************************/
/*********************************************************************************/

// Context: Update Transaction
//
// Access: Update
//
// Privileges: database authorization level

BdbStatus BdbDistribution::clusteringUnrecordDb (BdbDistDatabase& db)
{
    verbose ("Unregistering " + db.dbName() + " ...");

    // Check for the transaction mode
    if (!checkTransaction(BdbcUpdate))
        return error ("Need an update transaction");

    // Preliminary checks
    if (!db.isDbExist())
        return error ("No such database: " + db.dbName());

    // Privileges
    if (!db.isAuthorized())
        return error ("No privilege to manage: " + db.dbName());

    BdbDbTreeSingleton* treesingleton = BdbDbTreeSingleton::instance();
    BdbDbInfo dbinfo (db.dbName(), db.host(), db.filePath());

    if (BdbcSuccess != treesingleton->unregisterDb(dbinfo))
        return error ("Failed to unregister " + db.dbName());

    return BdbcSuccess;
}

/*********************************************************************************/

 

---

BaBar Public Site | SLAC | News | Links | Who's Who | Contact Us

Page Owner: Jacek Becla
Last Update: October 04, 2002