/*
 * Written by Dawid Kurzyniec and released to the public domain, as explained
 * at http://creativecommons.org/licenses/publicdomain
 */

package edu.emory.mathcs.util.io;

import java.io.*;
import edu.emory.mathcs.util.allocator.*;
import edu.emory.mathcs.backport.java.util.concurrent.*;

/**
 * In-memory pipe that enables buffered sequential data transfer between
 * threads. Pipe has two ends: source and sink. Source is an output stream
 * into which the writer writes bytes. Sink is the input stream from which
 * reader reads bytes. Data that was wrote to the source but not yet read from
 * the sink are kept in a pipe buffer.
 * <p>
 * This implementation features dynamic buffer sizing, so that memory
 * consumption is minimized if there is little data to buffer.
 * Buffer is upsized if more space is needed and downsized when data is read.
 * Resizing never causes data copying.
 * <p>
 * This implementation supports concurrent reads and writes.
 * <p>
 * Memory usage limits and allocation policy can be controlled via provided
 * {@link Allocator}. For example, multiple pipes can share allocator with
 * fixed maximum memory footprint, thus limiting total memory used for I/O
 * buffering.
 *
 * @see edu.emory.mathcs.util.net.inproc.InProcServerSocket
 * @see edu.emory.mathcs.util.net.inproc.InProcSocket
 *
 * @author Dawid Kurzyniec
 * @version 1.0
 */
public class BufferedPipe {

    /**
     * Default allocator with maximum memory footprint of 100 MB.
     */
    static Allocator defaultAllocator =
       new PoolingAllocator(50*1024*1024, 100*1024*1024, 2*1024*1024);

    static class Chunk {
        final Allocator.Buffer buf;
        final byte[] data;
        // after overflow, beg and end change sign (they become ~beg and ~end)
        // thanks to this trick, beg==end always indicate empty chunk, and
        // beg == ~end indicate full chunk
        volatile int beg, end;
        volatile Chunk next;
        Chunk(Allocator.Buffer buf) {
            this.buf = buf;
            // we will be using as much space as we can from this buffer
            // (it could be more than we requested). It is OK, since we do
            // not care about the random content of the array.
            this.data = buf.getData();
            this.beg = 0;
            this.end = 0;
        }
        void reclaim() {
            buf.releaseRef();
        }
    }

    final Allocator allocator;

    volatile Chunk begChunk;
    volatile Chunk endChunk;

    volatile boolean sourceClosed = false;
    volatile boolean sinkClosed = false;

    volatile int chunksize;

    //volatile boolean empty = true;
    final Object emptyLock = new Object();
    final Object fullLock = new Object();

    private class PipeRedirectibleInputStream extends InputStream
        implements TimedRedirectibleInput
    {
        public synchronized int read() throws IOException {
            try {
                return BufferedPipe.this.read(0);
            }
            catch (TimeoutException e) {
                throw new RuntimeException(e);
            }
        }
        public synchronized int timedRead(long timeout) throws IOException,
                                                               TimeoutException {
            return BufferedPipe.this.read(timeout);
        }
        public synchronized int read(byte[] buf, int off, int len) throws IOException {
            try {
                return BufferedPipe.this.read(buf, off, len, 0);
            }
            catch (TimeoutException e) {
                throw new RuntimeException(e);
            }
        }
        public synchronized int timedRead(byte[] buf, int off, int len, long timeout)
                throws IOException, TimeoutException {
            return BufferedPipe.this.read(buf, off, len, timeout);
        }
        public synchronized int timedRead(byte[] buf, long timeout) throws IOException,
                                                                           TimeoutException {
            return BufferedPipe.this.read(buf, 0, buf.length, timeout);
        }
        public synchronized int redirect(OutputStream dest, int len) throws IOException {
            try {
                return BufferedPipe.this.read(dest, len, 0);
            }
            catch (TimeoutException e) {
                throw new RuntimeException(e);
            }
        }
        public synchronized int timedRedirect(OutputStream dest, int len, long timeout)
            throws IOException, TimeoutException
        {
            return BufferedPipe.this.read(dest, len, timeout);
        }
        public synchronized int redirectAll(OutputStream dest) throws IOException {
            int total = 0;
            int read;
            try {
                while (true) {
                    read = BufferedPipe.this.read(dest, Integer.MAX_VALUE, 0);
                    if (read < 0)
                        return total;
                    total += read;
                }
            }
            catch (TimeoutException e) {
                throw new RuntimeException(e);
            }
        }
        public void close() {
            BufferedPipe.this.closeSink();
        }
    }

    /**
     * Source of the pipe.
     */
    protected final OutputStream source = new OutputStream() {
        public synchronized void write(int v) throws IOException {
            BufferedPipe.this.write(v);
        }
        public synchronized void write(byte[] buf, int off, int len) throws IOException {
            BufferedPipe.this.write(buf, off, len);
        }
        public void close() {
            BufferedPipe.this.closeSource();
        }
    };

    /**
     * Sink of the pipe.
     */
    protected final InputStream sink = new PipeRedirectibleInputStream();

    volatile Thread writerThread;

    /**
     * Creates a new pipe with a default shared allocator with 10 MB footprint
     * limit, and a default initial chunk size of 8 KB.
     */
    public BufferedPipe() {
        this(8192);
    }

    /**
     * Creates a new pipe with a default shared allocator with 10 MB footprint
     * limit, and with specified initial chunk size. Specifying large
     * initial chunk size increases initial memory footprint, but may slightly
     * improve initial performance of bulk data transfer.
     *
     * @param chunksize the initial chunk size.
     */
    public BufferedPipe(int chunksize) {
        this(defaultAllocator, chunksize);
    }

    /**
     * Creates a new pipe with specified allocator and a default initial chunk
     * size of 8 KB.
     * The allocator is used to provide memory buffers used by the pipe.
     *
     * @param allocator allocator that will provide memory buffers for this
     *                  pipe
     */
    public BufferedPipe(Allocator allocator) {
        this(allocator, 8192);
    }

    /**
     * Creates a new pipe with specified allocator and initial chunk size.
     * The allocator is used to provide memory buffers used by the pipe.
     * Specifying large initial chunk size increases initial memory footprint,
     * but may slightly improve performance for bulk data transfer.
     *
     * @param allocator allocator that will provide memory buffers for this
     *                  pipe
     * @param chunksize the initial chunk size
     */
    public BufferedPipe(Allocator allocator, int chunksize) {
        this.allocator = allocator;
        this.chunksize = chunksize;
    }

    /**
     * Returns the source of this pipe, where writer can write data.
     * @return the source of this pipe
     */
    public OutputStream source() {
        return source;
    }

    /**
     * Returns the sink of this pipe, from where reader can read data.
     * @return the sink of this pipe
     */
    public InputStream sink() {
        return sink;
    }

    /**
     * Returns the current length of the pipe, that is, the number of bytes
     * buffered and available for read. This is a snapshot value only.
     *
     * @return the current length of the pipe
     */
    public long length() {
        long length = 0;
        Chunk current = begChunk;
        while (current != null) {
            length += (current.end - current.beg);
            current = current.next;
        }
        return length;
    }


    int read(long timeout) throws IOException, TimeoutException {
        switch (ensureData(timeout)) {
            case EOF: return -1;
            case TIMEOUT: throw new TimeoutException("read timeout");
            default: break;
        }

        Chunk current = this.begChunk;
        byte[] data = current.data;
        int beg = current.beg;
        int end;

        // synchronize to ensure that all data is written back by the writer
        // before we accesses it
        synchronized (current) {
            end = current.end;
        }

        int v = current.data[beg++] & 0xff;

        if (beg < data.length) {
            current.beg = beg;
        }
        else {
            // end of chunk; remove and go to the next one
            Chunk next;
            next = current.next;
            if (next != null) {
                this.begChunk = next;
            }
            else {
                // possibly empty pipe; need to to make sure that reading
                // next and storing it to begChunk is atomic
                synchronized (emptyLock) {
                    next = current.next;
                    this.begChunk = next;
                }
            }

            // "current" now removed from the linked list; safe to reclaim
            current.reclaim();
        }

        return v;
    }

    int read(byte[] buf, int off, int len, long timeout) throws IOException,
                                                                TimeoutException {
        switch (ensureData(timeout)) {
            case EOF: return -1;
            case TIMEOUT: throw new TimeoutException("read timeout");
            default: break;
        }

        Chunk current = this.begChunk;
        int read_total = 0;

        while (true) {
            byte[] data = current.data;
            int beg = current.beg;
            int end;

            // synchronize to ensure that all data is written back by the writer
            // before we accesses it
            synchronized (current) {
                end = current.end;
            }

            int read_now = len < end-beg ? len : end-beg;

            //assert (read_now != 0);
            System.arraycopy(data, beg, buf, off, read_now);

            beg += read_now;
            off += read_now;
            len -= read_now;
            read_total += read_now;

            if (beg < data.length) {
                // done: no more data or no more requested
                current.beg = beg;
                return read_total;
            }
            else {
                // end of chunk; remove and go to the next one
                Chunk next;
                next = current.next;
                if (next != null) {
                    this.begChunk = next;
                }
                else {
                    // possibly empty pipe; need to to make sure that reading
                    // next and storing it to begChunk is atomic
                    synchronized (emptyLock) {
                        next = current.next;
                        this.begChunk = next;
                    }
                }

                // "current" now removed from the linked list; safe to reclaim
                current.reclaim();

                if (next == null) {
                    // all available data has been read
                    return read_total;
                }
                current = next;
            }
        }
    }

    int read(OutputStream os, int len, long timeout) throws IOException,
                                                            TimeoutException {
        switch (ensureData(timeout)) {
            case EOF: return -1;
            case TIMEOUT: throw new TimeoutException("read timeout");
            default: break;
        }

        Chunk current = this.begChunk;
        int read_total = 0;

        while (true) {
            byte[] data = current.data;
            int beg = current.beg;
            int end;

            // synchronize to ensure that all data is written back by the writer
            // before we accesses it
            synchronized (current) {
                end = current.end;
            }

            int read_now = len < end-beg ? len : end-beg;

            //assert (read_now != 0);
            os.write(data, beg, read_now);

            beg += read_now;
            len -= read_now;
            read_total += read_now;

            if (beg < data.length) {
                // done: no more data or no more requested
                current.beg = beg;
                return read_total;
            }
            else {
                // end of chunk; remove and go to the next one
                Chunk next;
                next = current.next;
                if (next != null) {
                    this.begChunk = next;
                }
                else {
                    // possibly empty pipe; need to to make sure that reading
                    // next and storing it to begChunk is atomic
                    synchronized (emptyLock) {
                        next = current.next;
                        this.begChunk = next;
                    }
                }

                // "current" now removed from the linked list; safe to reclaim
                current.reclaim();

                if (next == null) {
                    // all available data has been read
                    return read_total;
                }
                current = next;
            }
        }
    }

    void write(int v) throws IOException {
        // make sure there is space in the last chunk
        ensureSpace();

        Chunk current = this.endChunk;
        int end = current.end;
        current.data[end++] = (byte)v;

        // synchronize to ensure that all data is written back before
        // reader accesses it
        synchronized (current) {
            current.end = end;
        }

        // notify reader that buffer now is not empty
        synchronized (emptyLock) {
            if (this.begChunk == null) {
                this.begChunk = current;
            }
            emptyLock.notify();
        }
    }

    void write(byte[] buf, int off, int len) throws IOException {
        while (len > 0) {
            // make sure there is space in the last chunk
            ensureSpace();

            Chunk current = this.endChunk;
            int end = current.end;
            byte[] data = current.data;

            int write = (len < data.length-end ? len : data.length-end);
            System.arraycopy(buf, off, data, end, write);
            off += write;
            len -= write;
            end += write;

            // synchronize to ensure that all data is written back before
            // reader accesses it
            synchronized (current) {
                current.end = end;
            }

            // notify reader that buffer now is not empty
            synchronized (emptyLock) {
                if (this.begChunk == null) {
                    this.begChunk = current;
                }
                emptyLock.notify();
            }
        }
    }

    void closeSource() {
        sourceClosed = true;
        // notify reader that may wait on empty pipe that data will not arrive;
        // notify writer that may wait on full pipe that the source is closed
        fireIOStateChanged();
    }

    void closeSink() {
        sinkClosed = true;
        // notify reader that may wait on empty pipe that it is closed;
        // notify writer that may wait on full pipe that the pipe is broken
        fireIOStateChanged();
    }

    private void fireIOStateChanged() {
        synchronized (emptyLock) {
            emptyLock.notify();
        }
        synchronized (fullLock) {
            Thread writerThread = this.writerThread;
            if (writerThread != null) {
                writerThread.interrupt();
            }
        }
    }

    /**
     * Used only by read() operations.
     * @return flag indicating if pipe is empty
     */
    private boolean isEmpty() {
        Chunk begChunk = this.begChunk;
        // end of chunk, or there is a single chunk with beg == end
        return (begChunk == null || begChunk.beg == begChunk.end);
    }

    /**
     * Waits until there is at least one byte available for reading
     * in current chunk
     */
    private int ensureData(long timeout) throws IOException {
        // check if stream has not been closed
        if (sinkClosed) {
            throw new IOException("Stream closed");
        }
        if (!isEmpty()) {
            return DATA;
        }
        // empty pipe
        // synchronize to avoid being notified "non-empty" before waiting
        long tgt = (timeout <= 0 ? 0 : timeout + System.currentTimeMillis());
        synchronized (emptyLock) {
            while (isEmpty()) {
                // check if any of streams wasn't closed meanwhile
                if (sinkClosed) {
                    throw new IOException("Stream closed");
                }
                if (sourceClosed) {
                    return EOF;
                }
                try {
                    if (tgt == 0) {
                        emptyLock.wait();
                    }
                    else {
                        emptyLock.wait(timeout);
                        timeout = tgt - System.currentTimeMillis();
                        if (timeout <= 0) {
                            return TIMEOUT;
                        }
                    }
                }
                catch (InterruptedException e) {
                    throw newInterruptedIOException(e);
                }
            }
        }
        return DATA;
    }

    /**
     * Called only by write() operations
     */
    private void ensureSpace() throws IOException {
        checkWriteConsistency();

        Chunk chunk = this.endChunk;
        if (chunk != null && chunk.end < chunk.data.length) {
            // there is space in that chunk
            return;
        }

        Allocator.Buffer newbuf;
        try {

            // non-blocking attempt for double the size of the last chunk
            newbuf = allocator.allocate(chunksize, false, 0);

            if (newbuf == null) {
                // back up to half of planned size (equal to the last chunk)
                // and wait until completes
                synchronized (fullLock) {
                    checkWriteConsistency();
                    this.writerThread = Thread.currentThread();
                }
                try {
                    newbuf = allocator.allocate(chunksize/2, false, -1);
                } finally {
                    synchronized (fullLock) {
                        this.writerThread = null;
                    }
                }
            }
        }
        catch (InterruptedException e) {
            // first, make sure the interruption has not been induced by
            // asynchronous close
            checkWriteConsistency();
            // no, it is ordinary interruption
            throw newInterruptedIOException(e);
        }

        // store the planned size of next chunk to allocate
        this.chunksize = newbuf.getSize()*2;

        // append new chunk to the list
        Chunk next = new Chunk(newbuf);
        if (chunk != null) {
            chunk.next = next;
            this.endChunk = next;
        }
        else {
            this.begChunk = next;
            this.endChunk = next;
        }
    }

    private void checkWriteConsistency() throws IOException {
        if (sourceClosed) {
            throw new IOException("Stream closed");
        }
        if (sinkClosed) {
            throw new IOException("Broken pipe");
        }
    }

    private static InterruptedIOException newInterruptedIOException(InterruptedException e) {
        InterruptedIOException io = new InterruptedIOException(e.toString());
        //io.initCause(e);
        return io;
    }

    private final static int DATA    = 1;
    private final static int EOF     = 2;
    private final static int TIMEOUT = 3;
}
