path: root/src/main/java/org/gnunet/util/Helper.java

/*
 This file is part of GNUnet.
  Copyright (C) 2012, 2013 Christian Grothoff (and other contributing authors)

  GNUnet 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, or (at your
  option) any later version.

  GNUnet 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 GNUnet; see the file COPYING.  If not, write to the
  Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  Boston, MA 02111-1307, USA.
 */
package org.gnunet.util;


import org.gnunet.construct.Construct;
import org.gnunet.mq.Envelope;
import org.gnunet.mq.MessageQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOError;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.LinkedList;
import java.util.List;

/**
 * Process that we can communicate to with standard GNUnet messages over stdin/stdout.
 * The implementation uses threads, as Java does not support async I/O on process handles.
 */
public class Helper extends MessageQueue {
    private static final Logger logger = LoggerFactory
            .getLogger(Helper.class);

    private final ProcessBuilder processBuilder;
    private final RunaboutMessageReceiver receiver;
    private Process process;

    private WriteThread writeThread;
    private ReadThread readThread;

    private Scheduler.TaskConfiguration writeTaskConfig;
    private Scheduler.TaskIdentifier writeTaskId;

    private Scheduler.TaskConfiguration readTaskConfig;
    private Scheduler.TaskIdentifier readTaskId;

    private ByteBuffer writeBuffer = ByteBuffer.allocate(1024);

    private MessageStreamTokenizer mst;

    private boolean killed;


    /**
     * Thread that writes data from a pipe to the helper process.
     */
    private final class WriteThread extends Thread {
        final Pipe pipe;
        public WriteThread() {
            try {
                pipe = Pipe.open();
            } catch (IOException e) {
                throw new RuntimeException("JVM does not support pipes");
            }
        }
        @Override
        public void run() {
            WritableByteChannel channel = Channels.newChannel(process.getOutputStream());
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            while (true) {
                buffer.clear();
                try {
                    logger.debug("waiting for readable pipe in thread");
                    int n = pipe.source().read(buffer);
                    if (n == -1) {
                        logger.debug("write thread pipe closed");
                        return;
                    }
                    logger.debug("read {} bytes from pipe", n);
                    buffer.flip();
                    while (buffer.hasRemaining()) {
                        n = channel.write(buffer);
                        process.getOutputStream().flush();
                        logger.debug("written {} bytes to process", n);
                    }
                } catch (ClosedChannelException e) {
                    logger.debug("channel closed for write thread");
                    return;
                } catch (IOException e) {
                    throw new IOError(e);
                }
            }
        }
    }

    /**
     * Thread that reads data from the helper process and writes it
     * to a pipe.
     */
    private final class ReadThread extends Thread {
        Pipe pipe;

        public ReadThread() {
            try {
                pipe = Pipe.open();
            } catch (IOException e) {
                throw new RuntimeException("JVM does not support pipes");
            }
        }
        @Override
        public void run() {
            byte[] buf = new byte[1024];
            while (true) {
                int n;
                try {
                    logger.debug("reading from process");
                    n = process.getInputStream().read(buf);
                    if (n == -1) {
                        logger.debug("reached EOF while reading from process");
                        break;
                    }
                    logger.debug("read {} bytes from process", n);
                    ByteBuffer b = ByteBuffer.wrap(buf, 0, n);
                    while (b.hasRemaining()) {
                        pipe.sink().write(b);
                    }
                } catch (ClosedChannelException e) {
                    logger.debug("channel closed for read thread");
                    return;
                }
                catch (IOException e) {
                    throw new IOError(e);
                }
            }
        }
    }

    private String getBinaryPath(String binaryName) {
        if (!binaryName.contains("gnunet"))
            return binaryName;
        String prefix = System.getenv("GNUNET_PREFIX");
        if (prefix == null || prefix.isEmpty())
            return binaryName;
        // FIXME: this is not portable at all
        return prefix + "/lib/" + "gnunet/" + "libexec/" +  binaryName;
    }


    private class ReadTask implements Scheduler.Task {
        @Override
        public void run(Scheduler.RunContext ctx) {
            readTaskId = null;
            int n;
            try {
                n = mst.readFrom(readThread.pipe.source(), false);
            } catch (IOException e) {
                logger.warn("helper reader got io exception: {}", e);
                return;
            }
            if (n == -1) {
                logger.debug("helper reader got EOF");
                return;
            }
            if (!readThread.pipe.sink().isOpen()) {
                logger.debug("read thread closed pipe");
                return;
            }
            readTaskId = readTaskConfig.schedule();
        }
    }

    private class WriteTask implements Scheduler.Task {
        @Override
        public void run(Scheduler.RunContext ctx) {
            writeTaskId = null;
            try {
                int n = writeThread.pipe.sink().write(writeBuffer);
                logger.debug("wrote {} bytes in write task", n);
            } catch (IOException e) {
                throw new IOError(e);
            }
            if (writeBuffer.hasRemaining()) {
                writeTaskId = writeTaskConfig.schedule();
            } else {
                reportMessageSent();
                reportReadyForSubmit();
            }
        }
    }

    private class HelperMstCallback implements MstCalllback {
        @Override
        public void onUnknownMessage(UnknownMessageBody b) {
            logger.warn("got unknown message type");
        }

        @Override
        public void onKnownMessage(GnunetMessage msg) {
            logger.debug("processing message from helper");
            receiver.process(msg.body);
        }
    }

    /**
     * Create and start a new helper process
     *
     * @param withControlPipe (not implemented yet)
     * @param binaryName binary name of the helper process
     * @param argv arguments to the helper process
     * @param receiver receiver for messages from the process
     */
    public Helper(boolean withControlPipe, String binaryName, List<String> argv,
                  RunaboutMessageReceiver receiver) {
        logger.debug("in helper constructor");
        this.receiver = receiver;
        List<String> command = new LinkedList<String>();
        if (binaryName == null) {
            throw new AssertionError();
        }
        command.add(getBinaryPath(binaryName));
        if (argv != null)
            command.addAll(argv);
        processBuilder = new ProcessBuilder(command);
        try {
            process = processBuilder.start();
        } catch (IOException e) {
            throw new IOError(e);
        }

        mst = new MessageStreamTokenizer(new HelperMstCallback());

        writeThread = new WriteThread();
        readThread = new ReadThread();

        logger.debug("creating helper");

        try {
            readThread.pipe.source().configureBlocking(false);
            writeThread.pipe.sink().configureBlocking(false);
        } catch (IOException e) {
            throw new IOError(e);
        }

        writeThread.start();
        readThread.start();

        readTaskConfig = new Scheduler.TaskConfiguration(RelativeTime.FOREVER, new ReadTask());
        readTaskConfig.addSelectEvent(readThread.pipe.source(), SelectionKey.OP_READ);

        writeTaskConfig = new Scheduler.TaskConfiguration(RelativeTime.FOREVER, new WriteTask());
        writeTaskConfig.addSelectEvent(writeThread.pipe.sink(), SelectionKey.OP_WRITE);

        readTaskId = readTaskConfig.schedule();

        reportReadyForSubmit();
    }

    /**
     * Sends termination signal to the helper process.
     *
     * @param softkill if GNUNET_YES, signals termination by closing the helper's
     *          stdin; GNUNET_NO to signal termination by sending SIGTERM to helper
     * @return true on success, false on failure
     */
    public boolean kill(boolean softkill) {
        logger.debug("killing helper process (soft={})", softkill);
        killed = true;
        if (readTaskId != null) {
            readTaskId.cancel();
            readTaskId = null;
        }
        if (writeTaskId != null) {
            writeTaskId.cancel();
            writeTaskId = null;
        }

        try {
            readThread.pipe.source().close();
            readThread.pipe.sink().close();
            writeThread.pipe.source().close();
            writeThread.pipe.sink().close();
        } catch (IOException e) {
            throw new IOError(e);
        }
        if (softkill) {
            try {
                process.getInputStream().close();
            } catch (IOException e) {
                return false;
            }
            return true;
        }
        process.destroy();
        return true;
    }

    /**
     * Reap the helper process.  This call is blocking(!).  The helper process
     * should either be sent a termination signal before or should be dead before
     * calling this function
     *
     * @return true on success, false on failure
     */
    public boolean waitFor() {
        try {
            process.waitFor();
        } catch (InterruptedException e) {
            return false;
        }
        return true;
    }

    @Override
    protected void submit(Envelope ev) {
        if (writeTaskId != null) {
            throw new AssertionError();
        }
        logger.debug("submitting envelope to helper thread");
        final GnunetMessage gm = new GnunetMessage();
        gm.header = new GnunetMessage.Header();
        gm.body = ev.message;
        Construct.patch(gm);
        gm.header.messageSize = Construct.getSize(gm);
        if (writeBuffer.capacity() < gm.header.messageSize) {
            writeBuffer = ByteBuffer.allocate(gm.header.messageSize);
        }
        writeBuffer.clear();
        int n = Construct.write(writeBuffer, gm);
        writeBuffer.flip();
        logger.debug("wrote message size {} to write buffer", n);
        writeTaskId = writeTaskConfig.schedule();
    }

    @Override
    protected void retract() {
        throw new UnsupportedOperationException();
    }
}