/* This file is part of VoltDB.
 * Copyright (C) 2008-2012 VoltDB Inc.
 *
 * VoltDB is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * VoltDB 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 General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with VoltDB.  If not, see <http://www.gnu.org/licenses/>.
 */
package org.voltdb;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.channels.FileChannel;
import java.util.concurrent.Callable;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;

import org.voltcore.logging.VoltLogger;
import org.voltcore.utils.DBBPool.BBContainer;

import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.SettableFuture;

public class SimpleFileSnapshotDataTarget implements SnapshotDataTarget {
    private static final VoltLogger hostLog = new VoltLogger("HOST");

    private final File m_tempFile;
    private final File m_file;
    private final FileChannel m_fc;
    private long m_bytesWritten = 0;
    private Runnable m_onCloseTask;

    /*
     * If a write fails then this snapshot is hosed.
     * Set the flag so all writes return immediately. The system still
     * needs to scan all the tables to clear the dirty bits
     * so the process continues as if the writes are succeeding.
     * A more efficient failure mode would do the scan but not the
     * extra serialization work.
     */
    private volatile boolean m_writeFailed = false;
    private volatile Throwable m_writeException = null;

    public SimpleFileSnapshotDataTarget(
            File file) throws IOException {
        m_file = file;
        m_tempFile = new File(m_file.getParentFile(), m_file.getName() + ".incomplete");
        RandomAccessFile ras = new RandomAccessFile(m_tempFile, "rw");
        m_fc = ras.getChannel();

        m_es = MoreExecutors.listeningDecorator(
                Executors.newSingleThreadExecutor(new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {

                return new Thread(
                        Thread.currentThread().getThreadGroup(),
                        r,
                        "Snapshot write thread for " + m_file,
                        131072);
            }
        }));
    }

    private final ListeningExecutorService m_es;

    @Override
    public int getHeaderSize() {
        return 0;
    }

    @Override
    public ListenableFuture<?> write(final Callable<BBContainer> tupleData) {
        final SettableFuture<Object> retval = SettableFuture.create();
        final ListenableFuture<BBContainer> computedData = VoltDB.instance().getComputationService().submit(tupleData);
        computedData.addListener(new Runnable() {
            @Override
            public void run() {
                try {
                    final BBContainer data = computedData.get();
                    if (m_writeFailed) {
                        retval.set(null);
                        data.discard();
                        return;
                    }
                    try {
                        while (data.b.hasRemaining()) {
                            int written = m_fc.write(data.b);
                            if (written > 0) {
                                m_bytesWritten += written;
                            }
                        }
                    } finally {
                        data.discard();
                    }
                } catch (Throwable t) {
                    m_writeException = t;
                    hostLog.error("Error while attempting to write snapshot data to file " + m_file, t);
                    m_writeFailed = true;
                    retval.setException(t);
                    return;
                }
                retval.set(null);
            }
        }, m_es);
        return retval;
    }

    @Override
    public void close() throws IOException, InterruptedException {
        try {
            m_fc.force(false);
            m_fc.close();
            m_tempFile.renameTo(m_file);
        } finally {
            m_onCloseTask.run();
        }
    }

    @Override
    public long getBytesWritten() {
        return m_bytesWritten;
    }

    @Override
    public void setOnCloseHandler(Runnable onClose) {
        m_onCloseTask = onClose;
    }

    @Override
    public Throwable getLastWriteException() {
        return m_writeException;
    }

}

@@ -945,29 +945,155 @@
 current.
-ThreadFactory
+Future;%0Aimport java.util.concurrent.ThreadFactory;%0Aimport java.util.concurrent.TimeUnit;%0Aimport java.util.concurrent.atomic.AtomicReference
 ;%0A%0Aimpor
@@ -3255,22 +3255,16 @@
         
-final 
 Listenab
@@ -3361,16 +3361,282 @@
 eData);%0A
+        //Need to be able to null this out to prevent pack rat on the future%0A        //that wraps up the result of the write.%0A        final AtomicReference%3CFuture%3CBBContainer%3E%3E computedDataFinal =%0A            new AtomicReference%3CFuture%3CBBContainer%3E%3E(computedData);%0A%0A
         
@@ -3779,16 +3779,23 @@
   final 
+Future%3C
 BBContai
@@ -3801,28 +3801,138 @@
 iner
- data = computedData
+%3E fut = computedDataFinal.get();%0A                    computedDataFinal.set(null);%0A                    final BBContainer data = fut
 .get
@@ -5014,32 +5014,116 @@
 %7B%0A        try %7B%0A
+            m_es.shutdown();%0A            m_es.awaitTermination(356, TimeUnit.DAYS);%0A
             m_fc

Fix a pack rat bug in SimpleFileSnapshotTarget. Also fix race in close.