/*
 * Copyright 2012 University of Chicago
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */


package org.griphyn.vdl.mapping.file;

import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.log4j.Logger;
import org.griphyn.vdl.mapping.Mapper;
import org.griphyn.vdl.mapping.Path;
import org.griphyn.vdl.mapping.PhysicalFormat;
import org.griphyn.vdl.mapping.RootHandle;
import org.griphyn.vdl.util.SwiftConfig;

public class FileGarbageCollector implements Runnable {
    public static final Logger logger = Logger.getLogger(FileGarbageCollector.class);
    
    private static FileGarbageCollector instance;
    
    public synchronized static FileGarbageCollector getDefault() {
        if (instance == null) {
            instance = new FileGarbageCollector();
        }
        return instance;
    }
    
    private LinkedList<RootHandle> queue;
    private Thread thread;
    private Set<RootHandle> nodes;
    private boolean shutdown, done, enabled;
    
    public FileGarbageCollector() {
        queue = new LinkedList<RootHandle>();
        nodes = Collections.synchronizedSet(new HashSet<RootHandle>());
        enabled = SwiftConfig.getDefault().isFileGCEnabled();
        if (enabled) {
            thread = new Thread(this, "File Garbage Collector");
            thread.setDaemon(true);
            thread.start();
        }
    }
        
    public void clean(RootHandle node) {
        if (!enabled) {
            return;
        }
        nodes.remove(node);
        queue(node);
    }
    
    private void queue(RootHandle node) {
        synchronized(queue) {
            queue.add(node);
            queue.notify();
        }
    }

    public void add(RootHandle node) {
        if (!enabled) {
            return;
        }
        nodes.add(node);
    }
    
    public void clean() {
        if (!enabled) {
            return;
        }
        synchronized(queue) {
            queue.addAll(nodes);
            nodes.clear();
            queue.notify();
        }
    }
    
    public void run() {
        try {
            while (true) {
                RootHandle node;
                synchronized(queue) {
                    while (queue.isEmpty() && !shutdown) {
                        queue.wait();
                    }
                    if (shutdown && queue.isEmpty()) {
                        done = true;
                        break;
                    }
                    node = queue.remove();
                }
                try {
                    Mapper m = node.getMapper();
                    Collection<Path> fringe = node.getFringePaths();
                    for (Path p : fringe) {
                        PhysicalFormat pf = m.map(p);
                        if (!m.isPersistent(p)) {
                            pf.clean();
                            m.fileCleaned(pf);
                        }
                    }
                }
                catch (Exception e) {
                    logger.info("Failed to clean " + node, e);
                }
            }
        }
        catch (InterruptedException e) { 
        }
        done = true;
    }

    public void waitFor() throws InterruptedException {
        shutdown = true;
        while (!done && enabled) {
            synchronized(queue) {
                queue.notify();
            }
            Thread.sleep(1);
        }
    }
}

@@ -727,35 +727,8 @@
 et;%0A
-import java.util.Iterator;%0A
 impo
@@ -777,57 +777,8 @@
 Set;
-%0Aimport java.util.concurrent.LinkedBlockingQueue;
 %0A%0Aim
@@ -973,16 +973,71 @@
 Handle;%0A
+import org.griphyn.vdl.mapping.nodes.AbstractDataNode;%0A
 import o
@@ -2011,24 +2011,112 @@
  %7D%0A        %0A
+    /**%0A     * Cleans the specified handle. Also removes it from the collector.%0A     */%0A
     public v
@@ -2396,32 +2396,163 @@
        %7D%0A    %7D%0A%0A
+    /**%0A     * Registers a root handle with the collector. The handle will %0A     * be cleaned at some point in the future.%0A     */%0A
     public void 
@@ -2656,32 +2656,106 @@
 de);%0A    %7D%0A    %0A
+    /**%0A     * Cleans all handles registered with this collector.%0A     */%0A
     public void 
@@ -3432,32 +3432,146 @@
           try %7B%0A
+                    AbstractDataNode dn = (AbstractDataNode) node;%0A                    if (!dn.isCleaned()) %7B%0A    
                 
@@ -3599,24 +3599,28 @@
 etMapper();%0A
+    
             
@@ -3696,16 +3696,20 @@
         
+    
 for (Pat
@@ -3720,24 +3720,28 @@
 : fringe) %7B%0A
+    
             
@@ -3802,24 +3802,28 @@
             
+    
 if (!m.isPer
@@ -3832,24 +3832,28 @@
 stent(p)) %7B%0A
+    
             
@@ -3908,16 +3908,20 @@
         
+    
 m.fileCl
@@ -3931,16 +3931,46 @@
 ed(pf);%0A
+                            %7D%0A
         

Fix attempt for NPE (bug 1453)