clusterstore.java

Java的面向对象数据库系统的源代码

    protected Cluster giveMeAnUnlockedCluster(Permissions perms) throws IOException, ClassNotFoundException {
        return createANewEmptyAndUsableCluster(perms);
    }

    /**
     Associates the specified container with a cluster.

     Iff this method returns normally (without exception), the container is pinned and thus
     has to be unpinned.

     Iff this method returns normally (without exception), the container (and thus the cluster of the container)
     is write locked

     @param container Container to be registered with one cluster.
     */
    public void registerContainerAndLock(StorageObjectContainer container, Permissions perms, Transaction locker, int lockLevel) throws Exception {
        if (env.logWriter.hasTarget(LogWriter.DEBUG3)) {
            env.logWriter.newEntry(this, "registerContainer()", LogWriter.DEBUG3);
        }

        Cluster cluster = null;

        boolean pinned = false;
        boolean locked = false;
        boolean alright = false;

        try {
            synchronized (this) {
                cluster = growingCluster(perms);

                Lock clusterLock = cluster.lock();
                int prevLevel = clusterLock.tryAcquire(locker, lockLevel);

                if (prevLevel == Lock.NOT_ACQUIRED) { // The cluster we are trying to lock is already locked, so we take another cluster
                    cluster = giveMeAnUnlockedCluster(perms);

                    clusterLock = cluster.lock();
                    prevLevel = clusterLock.tryAcquire(locker, lockLevel);

                    if (prevLevel == Lock.NOT_ACQUIRED) {
                        throw new Error("BUG! We could not acquire a lock for an unlocked cluster.");
                    }
                }
                locked = true;

                cluster.registerContainer(container);
                container.pin();
                pinned = true;
            }
            cluster.updateLockLevel(locker);

            if (env.logWriter.hasTarget(LogWriter.DEBUG3)) {
                env.logWriter.newEntry(this, "    cluster: " + cluster.clusterID(), LogWriter.DEBUG3);
            }
            alright = true;
        } finally {
            if (!alright) {
                if (locked) {
                    cluster.lock().release(locker);
                }
                if (pinned) {
                    container.unpin();
                }
            }
        }
    }


    public void invalidateContainer(StorageObjectContainer container) /*throws Exception*/ {
        synchronized (container) {
            container.getCluster().removeContainer(container);
            container.setCluster(null);
        }
    }


    protected Cluster restoreCluster(ClusterID cid) throws Exception {
        String basename = basename(cid);
        Cluster cluster;

        new File(basename + POSTFIX_LOCK).delete();
        new File(basename + POSTFIX_TEMP).delete();

        File shadowFile = new File(basename + POSTFIX_SHADOW);
        File clusterFile = new File(basename + POSTFIX_CLUSTER);

        if (shadowFile.exists()) {
            /*
                FIXME:
                Who says that shadow files are always better than cluster files?
                Ozone may have crashed just when starting to write the shadow file.
                The following if clause catches this if the file size is 0.
                But what if the file is written incompletely, but it's length is not zero?
                It will be regarded as intact copy, while it is not.

                Maybe we should return to atomic rename at WizardCluster.saveShadow()
            */
            if (shadowFile.length() > 0) {
                if (!shadowFile.renameTo(clusterFile)) {
                    throw new IOException("Unable to rename shadow file.");
                }
            } else {
                shadowFile.delete();
            }
        }
        cluster = (Cluster) loadData(basename + POSTFIX_CLUSTER);
        activateCluster(cluster, 0);

        return cluster;
    }


    /**
     * Make sure the corresponding cluster is in the cache. While loading
     * clusters, we may have to throw away (and maybe store) some currently
     * cached clusters.
     *
     *
     * @param cid ClusterID of the cluster to load.
     * @param pin
     wether the loaded cluster should be pinned as soon as it is loaded
     so that there may be no chance to unload unless it is unpinned.
     If this parameter is set to true, the user has to unpin the cluster.
     If this parameter is set to false, the cluster may already be unloaded when this method returns.
     after using it.
     */
    public Cluster loadCluster(ClusterID cid, boolean pin) throws IOException, ClassNotFoundException {
        Cluster cluster = (Cluster) cachedClusters.elementForKey(cid);
        if (cluster == null) {

            if (env.logWriter.hasTarget(LogWriter.DEBUG)) {
                env.logWriter.newEntry(this, "loadCluster(): load cluster from disk: " + cid.toString(), LogWriter.DEBUG);
            }

            String basename = basename(cid);
            String clusterName = basename + POSTFIX_CLUSTER;
            String lockName = basename + POSTFIX_LOCK;

            int clusterByteSize = (int) new File(clusterName).length();
            if (compressClusters) {
                clusterByteSize *= compressionFactor;
            }

            // make sure that many different threads don't load
            // to much data before the currently synchronized thread
            // can trim the cache
            trimCache();
            cluster = (Cluster) loadData(clusterName);

            synchronized (this) {

                // now we have to check the cachedClusters table inside the
                // synchronized block to see if someone did register this
                // cluster while we loaded it
                Cluster interimCluster = (Cluster) cachedClusters.elementForKey(cid);
                if (interimCluster != null) {
                    //with appropriate locking|pinning, it is no problem or danger.
                    env.logWriter.newEntry(this, "loadCluster(): cluster was loaded by another thread too; droping my copy", LogWriter.DEBUG);

                    cluster = interimCluster;

                    if (pin && cluster instanceof WizardCluster) {
                        ((WizardCluster)cluster).pin();
                    }
                    
                } else {
                    // we are going to mess with the cluster; it seems that the cluster
                    // is not visible to other thread until it is added to cachedClusters,
                    // however, IBM jdk throws an exception in cluster.updateLockLevel, which
                    // seems to be related to the initialization in the following block
                    synchronized (cluster) {
                        // locks are only there if the lock level is >= READ
                        try {
                            cluster.setLock((Lock) loadData(lockName));
                            new File(lockName).delete();
                            ((MROWLock) cluster.lock()).setDebugInfo("clusterID=" + cluster.clusterID());
                        } catch (Exception e) {
                            if (env.logWriter.hasTarget(LogWriter.DEBUG3)) {
                                env.logWriter.newEntry(this, "    Unable to load lock from disk - creating a new lock.", LogWriter.DEBUG3);
                            }
                            cluster.setLock(env.transactionManager.newLock());
                            ((MROWLock) cluster.lock()).setDebugInfo("clusterID=" + cluster.clusterID());
                        }

                        if (pin && cluster instanceof WizardCluster) { // We pin inside the synchronization to the cluster, because calling pin() will try another synchronization and two nested synchronizations to an object are faster than two serial synchronizations.
                            ((WizardCluster)cluster).pin();
                        }

                        activateCluster(cluster, clusterByteSize);
                    }

                    if (clusterByteSize > maxClusterSize * 2) {
                        splitCluster(cluster);
                    }

                    cachedClusters.addForKey(cluster, cluster.clusterID());

                    trimCache();
                }
            }
        } else {
            synchronized (cluster) {
                if (pin && cluster instanceof WizardCluster) {
                    ((WizardCluster)cluster).pin();
                }
            }
        }
        if (env.logWriter.hasTarget(LogWriter.DEBUG3)) {
            env.logWriter.newEntry(this, "returning WizardCluster: " + cluster, LogWriter.DEBUG3);
        }
        return cluster;
    }


    public void splitCluster(Cluster cluster) {
    }


    /**
     * Remove cluster from the cluster cache.
     * @param cid
     */
    public void unloadCluster(ClusterID cid, boolean deactivate) throws IOException {
        if (env.logWriter.hasTarget(LogWriter.DEBUG)) {
            env.logWriter.newEntry(this, "unloadCluster(" + cid + "," + deactivate + ").", LogWriter.DEBUG);
        }
        
        Cluster cluster = (Cluster) cachedClusters.removeForKey(cid);

        if (deactivate) {
            deactivateCluster(cluster);
        }
    }


    /**
     * Ensure that there is at least the specified size of free space in the
     * cluster cache. Under some circumstances clusters (currently invoked)
     * cannot be deactivated. Therefore this method cannot guarantee that the
     * needed space is free afterwards.<p>
     *
     * This is the central method of the deactivation of containers that are
     * currently in use. This is different from the commit behaviour.
     */
    protected void trimCache() throws IOException {

        long freeSpace = env.freeMemory();
        if (false && env.logWriter.hasTarget(LogWriter.DEBUG)) {
            env.logWriter.newEntry(this, "trimCache():  free:" + freeSpace, LogWriter.DEBUG2);
        }

        boolean tryRemoveCluster = true;
        while (freeSpace <= 0 && tryRemoveCluster) {
            tryRemoveCluster = false;
            synchronized (this) {
                long cacheSize = 0;

                // build priority queue for all currently loaded clusters
                DxMap priorityQueue = new DxTreeMap();
                DxIterator it = cachedClusters.iterator();
                Cluster cluster;
                while ((cluster = (Cluster) it.next()) != null) {
                    priorityQueue.addForKey(cluster, cluster.cachePriority());
                    cacheSize += cluster.size();
                }

                // free at least 20% of the cache
                long cacheSizeToRemove = cacheSize / 5;
                if (env.logWriter.hasTarget(LogWriter.DEBUG)) {
                    env.logWriter.newEntry(this, "   cache: " + cacheSize + " to be freed:" + cacheSizeToRemove, LogWriter.DEBUG2);
                }

                // throw away (deactivate) clusters, lowest priority first
                it = priorityQueue.iterator();
                while (cacheSizeToRemove > 0 && (cluster = (WizardCluster) it.next()) != null) {

                    // if any of the containers is currently invoked, the cluster
                    // must not be written and must stay in memory
                    // The same applies for pinned containers.
                    // FIXME: Once pinning is fully established, we may not need to call isInvoked() anymore.
                    if (cluster instanceof WizardCluster) {
                        WizardCluster wizardCluster = (WizardCluster) cluster;
                        if ((!wizardCluster.isPinned()) && (!wizardCluster.isInvoked())) {
                            if (env.logWriter.hasTarget(LogWriter.DEBUG)) {
                                env.logWriter.newEntry(this, "DEACTIVATE cluster: " + cluster.clusterID(), LogWriter.DEBUG2);
                            }

                            cluster = (Cluster) it.removeObject();
                            cacheSizeToRemove -= cluster.size();
                            unloadCluster(cluster.clusterID(), true);
                            tryRemoveCluster = true;
    //                        if (env.logWriter.hasTarget(LogWriter.DEBUG)) {
    //                            env.logWriter.newEntry(this, "   free:" + freeSpace, LogWriter.DEBUG2);