path: root/src/org/gnunet/statistics/Statistics.java

/*
 This file is part of GNUnet.
 (C) 2011, 2012 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.
 */

/*
 * The stuff below does nothing whatsoever, first milestone of
 * this project is to implement the StatisticsService api
 * 
 */

package org.gnunet.statistics;

import org.gnunet.requests.Request;
import org.gnunet.requests.RequestQueue;
import org.gnunet.util.*;
import org.gnunet.util.getopt.Argument;
import org.gnunet.util.getopt.ArgumentAction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;

/**
 * API for the gnunet statistics service.
 * <p/>
 * Set, get and monitor statistics values, represented as unsigned 64bit integer.
 * Note that {@literal long}, java's largest primitive type, can only store signed 64bit integers.
 * With absolute operation, its negative values are interpreted as large numbers by the statistics api.
 */
public class Statistics {
    private static final Logger logger = LoggerFactory
            .getLogger(Statistics.class);

    /**
     * Time after we give up on setting values in statistics
     */
    private static final RelativeTime SET_TIMEOUT = RelativeTime.SECOND.multiply(10);

    private final static int SETFLAG_RELATIVE = 1;
    private final static int SETFLAG_PERSIST = 2;

    private final Client client;

    private final RequestQueue requestQueue;

    /**
     * Callback for the current get request. Only one get request ist allowed at a time.
     */
    private StatisticsReceiver currentGetReceiver;
    /**
     * Success/Error continuation for the current get request.
     */
    private Continuation currentGetContinuation;

    /**
     * List of all watch requests, canceled watch requests are null (but stay due to protocol limitations)
     */
    private ArrayList<StatisticsWatchRequest> watchRequests = new ArrayList<StatisticsWatchRequest>();

    /**
     * A request to the statistics service.
     */
    private abstract class StatisticsRequest extends Request {
        public String name;
        public String subsystem;
        public AbsoluteTime deadline;
    }

    private class StatisticsGetRequest extends StatisticsRequest {
        public StatisticsReceiver receiver;

        public void onCancel(boolean alreadyTransmitted) {
            currentGetReceiver = null;
        }

        public AbsoluteTime getDeadline() {
            return deadline;
        }


        public void transmit(Connection.MessageSink sink) {
            GetMessage rm = new GetMessage();
            rm.statisticsName = name;
            rm.subsystemName = subsystem;

            sink.send(rm);
        }

        public boolean onReconnect() {
            return true;
        }
    }

    private class StatisticsSetRequest extends StatisticsRequest {
        public long value;
        public int flags;

        public AbsoluteTime getDeadline() {
            return SET_TIMEOUT.toAbsolute();
        }

        public void transmit(Connection.MessageSink sink) {
            SetMessage sm = new SetMessage();
            sm.statisticName = name;
            sm.subsystemName = subsystem;
            sm.value = value;
            sm.flags = flags;
            sink.send(sm);
        }

        public boolean onDestroy() {
            // keep the request
            return true;
        }

        public boolean onReconnect() {
            // just keep the request on reconnect
            return true;
        }
    }


    private class StatisticsWatchRequest extends StatisticsRequest {
        public StatisticsReceiver receiver;

        public AbsoluteTime getDeadline() {
            return AbsoluteTime.FOREVER;
        }

        public void onCancel(boolean alreadyTransmitted) {
            System.out.println("already transmitted: " + alreadyTransmitted);
            if (alreadyTransmitted) {
                watchRequests.set(watchRequests.indexOf(this), null);
            }

        }

        public boolean onReconnect() {
            // do this because we'll probably get new WatchIDs for every watch request on reconnect
            if (watchRequests.contains(this)) {
                watchRequests.clear();
            }

            watchRequests.add(this);

            return true;
        }

        public void transmit(Connection.MessageSink sink) {
            WatchMessage wm = new WatchMessage();
            wm.statisticsName = name;
            wm.subsystemName = subsystem;
            sink.send(wm);


            watchRequests.add(this);
        }
    }

    private static class TESTRequest extends Request {
        private AbsoluteTime deadline;

        public TESTRequest(AbsoluteTime deadline) {
            this.deadline = deadline;
        }

        public boolean onDestroy() {
            // keep on destroy
            return true;
        }

        public AbsoluteTime getDeadline() {
            return deadline;
        }

        public void transmit(Connection.MessageSink sink) {
            sink.send(new TestMessage());
            // todo: disconnect when not receiving the TEST message back after timeout
        }
    }


    public class StatisticsMessageReceiver extends RunaboutMessageReceiver {
        public void visit(GetResponseMessage m) {
	    currentGetReceiver.onReceive(m.subsystemName, m.statisticName, m.value);
        }

        public void visit(GetResponseEndMessage m) {
            currentGetReceiver = null;
            if (currentGetContinuation != null) {
                currentGetContinuation.cont(true);
            }
        }

        public void visit(TestMessage m) {
            client.disconnect();
        }

        public void visit(WatchResponseMessage wrm) {
            if (watchRequests.size() <= wrm.wid) {
                logger.warn("statistics service got confused with watch request");
                return;
            }
            StatisticsWatchRequest wr = watchRequests.get(wrm.wid);
            // request may have been canceled by the api (but not by the server)
            if (wr != null) {
                wr.receiver.onReceive(wr.subsystem, wr.name, wrm.value);
            }
        }

        @Override
        public void handleError() {
            requestQueue.reconnect();
        }
    }

    public Statistics(Configuration cfg) {
        client = new Client("statistics", cfg);
        requestQueue = new RequestQueue(this.client, new StatisticsMessageReceiver());
    }

    /**
     * Retrieve values from statistics.
     *
     * @param timeout      time after we give up and call receiver.onTimeout
     * @param subsystem    the subsystem of interest
     * @param name         name of the statistics value belongs to
     * @param receiver     callback
     * @param continuation
     * @return handle to cancel the request
     */
    public Cancelable get(RelativeTime timeout, final String subsystem, final String name,
                          final StatisticsReceiver receiver, Continuation continuation) {

        if (currentGetReceiver != null) {
            throw new AssertionError("only one Statistics get request can be active at a time");
        }
        currentGetReceiver = receiver;
        currentGetContinuation = continuation;

        final StatisticsGetRequest getRequest = new StatisticsGetRequest();
        getRequest.deadline = timeout.toAbsolute();
        getRequest.name = (name == null) ? "" : name;
        getRequest.subsystem = (subsystem == null) ? "" : subsystem;
        getRequest.receiver = receiver;

        return requestQueue.add(getRequest);
    }


    /**
     * Sets a statistics value asynchronously.
     *
     * @param name    name of the entry
     * @param value   desired value
     * @param persist keep value even if the statistics service restarts
     * @return a handle to cancel the request
     */
    public Cancelable set(final String subsystem, final String name, final long value, boolean persist) {
        StatisticsSetRequest setRequest = new StatisticsSetRequest();
        setRequest.deadline = SET_TIMEOUT.toAbsolute();
        setRequest.subsystem = subsystem;
        setRequest.name = name;
        setRequest.value = value;
        setRequest.flags = persist ? SETFLAG_PERSIST : 0;

        return requestQueue.add(setRequest);
    }

    /**
     * Changes a statistics value asynchronously.
     *
     * @param name    name of the entry
     * @param delta   relative difference to the old value
     * @param persist keep value even if the statistics service restarts
     * @return a handle to cancel the request
     */
    public Cancelable update(final String subsystem, final String name, final long delta, boolean persist) {
        StatisticsSetRequest setRequest = new StatisticsSetRequest();
        setRequest.deadline = SET_TIMEOUT.toAbsolute();
        setRequest.subsystem = subsystem;
        setRequest.name = name;
        setRequest.value = delta;
        setRequest.flags = (persist ? SETFLAG_PERSIST : 0) | SETFLAG_RELATIVE;

        return requestQueue.add(setRequest);
    }

    public Cancelable watch(final String subsystem, final String name, StatisticsReceiver receiver) {
        StatisticsWatchRequest watchRequest = new StatisticsWatchRequest();
        watchRequest.deadline = AbsoluteTime.FOREVER;
        watchRequest.name = name;
        watchRequest.subsystem = subsystem;
        watchRequest.receiver = receiver;

        // even after the request has been sent, we want to keep it
        // (e.g. for retransmission on reconnect)
        return requestQueue.addPersistent(watchRequest);
    }

    /**
     * Destroy handle to the statistics service. Always finishes writing pending values.
     */
    public void destroy() {
        // the request queue handles the destruction, maybe we still have important messages pending etc.
        requestQueue.add(new TESTRequest(SET_TIMEOUT.toAbsolute()));
        requestQueue.shutdown();
    }


    /**
     * Statistics command line utility entry point
     *
     * @param args command line arguments
     */
    public static void main(String[] args) {
        new Program(args) {
            @Argument(
                    shortname = "x",
                    longname = "set",
                    action = ArgumentAction.SET,
                    description = "set a value")
            boolean test;
            @Argument(
                    shortname = "w",
                    longname = "watch",
                    action = ArgumentAction.SET,
                    description = "set a value")
            boolean watch;
            @Argument(
                    shortname = "n",
                    longname = "name",
                    action = ArgumentAction.STORE_STRING,
                    description = "statistics name")
            String statisticsName = "";
            @Argument(
                    shortname = "s",
                    longname = "subsystem",
                    action = ArgumentAction.STORE_STRING,
                    description = "subsystem name")
            String subsystemName = "";

            public void run() {
                final Statistics statistics = new Statistics(cfg);
                if (test) {
                    if (subsystemName.isEmpty() || statisticsName.isEmpty()) {
                        System.err.println("must specify non-empty subsystem and name");
                        return;
                    }
                    if (unprocessedArgs.length != 1) {
                        System.err.println("must specify exactly one value to set");
                        return;
                    }
                    long value;
                    try {
                        value = Long.parseLong(unprocessedArgs[0]);
                    } catch (NumberFormatException e) {
                        System.err.println("invalid value (not a long)");
                        return;
                    }
                    statistics.set(subsystemName, statisticsName, value, false);
                    statistics.destroy();
                } else {
                    if (unprocessedArgs.length == 0) {
                        if (watch) {
                            statistics.watch(subsystemName, statisticsName,
                                    new StatisticsReceiver() {
                                        @Override
                                        public void onReceive(String subsystem, String name, long value) {
                                            System.out.println(subsystem + "(" + name + ") = " + value);
                                        }
                                    }
                            );
                        } else {
                            statistics.get(RelativeTime.SECOND, subsystemName, statisticsName,
                                    new StatisticsReceiver() {
                                        @Override
                                        public void onReceive(String subsystem, String name, long value) {
                                            System.out.println(subsystem + "(" + name + ") = " + value);
                                        }
                                    },
                                    new Continuation() {
                                        @Override
                                        public void cont(boolean success) {
                                            statistics.destroy();
                                        }
                                    }
                            );
                        }
                    } else {
                        System.err.println("dumping statistics does not take any positional parameters");
                    }
                }
            }
        }.start();
    }

}