/*************************************************************************

"FreePastry" Peer-to-Peer Application Development Substrate 

Copyright 2002, Rice University. All rights reserved.

Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions are
met:

- Redistributions of source code must retain the above copyright
notice, this list of conditions and the following disclaimer.

- Redistributions in binary form must reproduce the above copyright
notice, this list of conditions and the following disclaimer in the
documentation and/or other materials provided with the distribution.

- Neither  the name  of Rice  University (RICE) nor  the names  of its
contributors may be  used to endorse or promote  products derived from
this software without specific prior written permission.

This software is provided by RICE and the contributors on an "as is"
basis, without any representations or warranties of any kind, express
or implied including, but not limited to, representations or
warranties of non-infringement, merchantability or fitness for a
particular purpose. In no event shall RICE or contributors be liable
for any direct, indirect, incidental, special, exemplary, or
consequential damages (including, but not limited to, procurement of
substitute goods or services; loss of use, data, or profits; or
business interruption) however caused and on any theory of liability,
whether in contract, strict liability, or tort (including negligence
or otherwise) arising in any way out of the use of this software, even
if advised of the possibility of such damage.

********************************************************************************/

package rice.p2p.past;

import java.io.*;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.logging.*;

import rice.*;
import rice.Continuation.*;
import rice.environment.Environment;
import rice.environment.logging.Logger;
import rice.environment.params.Parameters;
import rice.p2p.commonapi.*;
import rice.p2p.commonapi.appsocket.*;
import rice.p2p.commonapi.rawserialization.*;
import rice.p2p.past.PastPolicy.*;
import rice.p2p.past.messaging.*;
import rice.p2p.past.rawserialization.*;
import rice.p2p.replication.*;
import rice.p2p.replication.manager.*;
import rice.p2p.util.MathUtils;
import rice.p2p.util.rawserialization.*;
import rice.persistence.*;

/**
 * @(#) PastImpl.java This is an implementation of the Past interface.
 *
 * @version $Id: PastImpl.java 3329 2006-06-27 13:02:52Z jstewart $
 * @author Alan Mislove
 * @author Ansley Post
 * @author Peter Druschel
 */
public class PastImpl implements Past, Application, ReplicationManagerClient {

  // ----- STATIC FIELDS -----
  // the number of milliseconds to wait before declaring a message lost
  /**
   * DESCRIBE THE FIELD
   */
  public final int MESSAGE_TIMEOUT;
  // = 30000;

  // the percentage of successful replica inserts in order to declare success
    /**
   * DESCRIBE THE FIELD
   */
  public final double SUCCESSFUL_INSERT_THRESHOLD;
  // = 0.5;

  // ----- VARIABLE FIELDS -----

  // this application's endpoint
    /**
   * DESCRIBE THE FIELD
   */
  protected Endpoint endpoint;

  // the storage manager used by this Past
  /**
   * DESCRIBE THE FIELD
   */
  protected StorageManager storage;

  // The trash can, or where objects should go once removed.  If null, they are deleted
  /**
   * DESCRIBE THE FIELD
   */
  protected StorageManager trash;

  // The backup store, or location of over-replicated objects, helping PAST to better deal with churn
  /**
   * DESCRIBE THE FIELD
   */
  protected Cache backup;

  // the replication factor for Past
  /**
   * DESCRIBE THE FIELD
   */
  protected int replicationFactor;

  // the replica manager used by Past
  /**
   * DESCRIBE THE FIELD
   */
  protected ReplicationManager replicaManager;

  /**
   * DESCRIBE THE FIELD
   */
  protected LockManager lockManager;

  // the policy used for application-specific behavior
  /**
   * DESCRIBE THE FIELD
   */
  protected PastPolicy policy;

  // the unique ids used by the messages sent across the wire
  private int id;

  // the hashtable of outstanding messages
  private Hashtable outstanding;

  // the hashtable of outstanding timer tasks
  private Hashtable timers;

  // the factory for manipulating ids
  /**
   * DESCRIBE THE FIELD
   */
  protected IdFactory factory;

  // the instance name we are running with
  /**
   * DESCRIBE THE FIELD
   */
  protected String instance;

  // debug variables
  /**
   * DESCRIBE THE FIELD
   */
  public int inserts = 0;
  /**
   * DESCRIBE THE FIELD
   */
  public int lookups = 0;
  /**
   * DESCRIBE THE FIELD
   */
  public int fetchHandles = 0;
  /**
   * DESCRIBE THE FIELD
   */
  public int other = 0;

  /**
   * DESCRIBE THE FIELD
   */
  protected Environment environment;
  /**
   * DESCRIBE THE FIELD
   */
  protected Logger logger;

  /**
   * DESCRIBE THE FIELD
   */
  protected PastContentDeserializer contentDeserializer;
  /**
   * DESCRIBE THE FIELD
   */
  protected PastContentHandleDeserializer contentHandleDeserializer;

  /**
   * DESCRIBE THE FIELD
   */
  public SocketStrategy socketStrategy;

  /**
   * AppSocket -> ByteBuffer[] Used for receiving the objects.
   */
  WeakHashMap pendingSocketTransactions = new WeakHashMap();

  /**
   * Constructor for Past, using the default policy
   *
   * @param node The node below this Past implementation
   * @param manager The storage manager to be used by Past
   * @param replicas The number of object replicas
   * @param instance The unique instance name of this Past
   */
  public PastImpl(Node node, StorageManager manager, int replicas, String instance) {
    this(node, manager, replicas, instance, new DefaultPastPolicy());
  }

  /**
   * Constructor for Past
   *
   * @param node The node below this Past implementation
   * @param manager The storage manager to be used by Past
   * @param replicas The number of object replicas
   * @param instance The unique instance name of this Past
   * @param policy DESCRIBE THE PARAMETER
   */
  public PastImpl(Node node, StorageManager manager, int replicas, String instance, PastPolicy policy) {
    this(node, manager, null, replicas, instance, policy, null);
  }

  /**
   * Constructor for PastImpl.
   *
   * @param node DESCRIBE THE PARAMETER
   * @param manager DESCRIBE THE PARAMETER
   * @param backup DESCRIBE THE PARAMETER
   * @param replicas DESCRIBE THE PARAMETER
   * @param instance DESCRIBE THE PARAMETER
   * @param policy DESCRIBE THE PARAMETER
   * @param trash DESCRIBE THE PARAMETER
   */
  public PastImpl(Node node, StorageManager manager, Cache backup, int replicas, String instance, PastPolicy policy, StorageManager trash) {
    this(node, manager, backup, replicas, instance, policy, trash, false);
  }

  /**
   * @param node
   * @param manager
   * @param backup
   * @param replicas
   * @param instance
   * @param policy
   * @param trash
   * @param useOwnSocket send all inserts/fetches over a socket (default is
   *      false)
   */
  public PastImpl(Node node, StorageManager manager, Cache backup, int replicas, String instance, PastPolicy policy, StorageManager trash, boolean useOwnSocket) {
    this(node, manager, backup, replicas, instance, policy, trash, new DefaultSocketStrategy(useOwnSocket));
  }

  /**
   * Constructor for Past
   *
   * @param node The node below this Past implementation
   * @param manager The storage manager to be used by Past
   * @param replicas The number of object replicas
   * @param instance The unique instance name of this Past
   * @param backup DESCRIBE THE PARAMETER
   * @param policy DESCRIBE THE PARAMETER
   * @param trash DESCRIBE THE PARAMETER
   * @param strategy DESCRIBE THE PARAMETER
   */
  public PastImpl(Node node, StorageManager manager, Cache backup, int replicas, String instance, PastPolicy policy, StorageManager trash, SocketStrategy strategy) {
    this.environment = node.getEnvironment();
    logger = environment.getLogManager().getLogger(getClass(), instance);
    Parameters p = environment.getParameters();
    MESSAGE_TIMEOUT = p.getInt("p2p_past_messageTimeout");
    // = 30000;
    SUCCESSFUL_INSERT_THRESHOLD = p.getDouble("p2p_past_successfulInsertThreshold");
    // = 0.5;
    this.socketStrategy = strategy;
    this.storage = manager;
    this.backup = backup;
    this.contentDeserializer = new JavaPastContentDeserializer();
    this.contentHandleDeserializer = new JavaPastContentHandleDeserializer();
    this.endpoint = node.buildEndpoint(this, instance);
    this.endpoint.setDeserializer(new PastDeserializer());
    this.factory = node.getIdFactory();
    this.policy = policy;
    this.instance = instance;
    this.trash = trash;

    this.id = Integer.MIN_VALUE;
    this.outstanding = new Hashtable();
    this.timers = new Hashtable();
    this.replicationFactor = replicas;

    //   log.addHandler(new ConsoleHandler());
    //   log.setLevel(Level.FINE);
    //   log.getHandlers()[0].setLevel(Level.FINE);

    this.replicaManager = buildReplicationManager(node, instance);

    this.lockManager = new LockManagerImpl(environment);

    this.endpoint.accept(
      new AppSocketReceiver() {

        public void receiveSocket(AppSocket socket) {
          if (logger.level <= Logger.FINE) {
            logger.log("Received Socket from " + socket);
          }
          socket.register(true, false, 10000, this);
          endpoint.accept(this);
        }

        public void receiveSelectResult(AppSocket socket, boolean canRead,
                                        boolean canWrite) {
          if (logger.level <= Logger.FINER) {
            logger.log("Reading from " + socket);
          }
          try {
            ByteBuffer[] bb = (ByteBuffer[]) pendingSocketTransactions.get(socket);
            if (bb == null) {
              // this is a new message

              // read the size
              bb = new ByteBuffer[1];
              bb[0] = ByteBuffer.allocate(4);
              if (socket.read(bb, 0, 1) == -1) {
                close(socket);
                return;
              }

              // TODO: need to handle the condition where we don't read the whole size...
              byte[] sizeArr = bb[0].array();
              int size = MathUtils.byteArrayToInt(sizeArr);

              if (logger.level <= Logger.FINER) {
                logger.log("Found object of size " + size + " from " + socket);
              }

              // allocate a buffer to store the object, save it in the pst
              bb[0] = ByteBuffer.allocate(size);
              pendingSocketTransactions.put(socket, bb);
            }

            // now we have a bb

            // read some bytes
            if (socket.read(bb, 0, 1) == -1) {
              close(socket);
            }

            // deserialize or reregister
            if (bb[0].remaining() == 0) {
              // make sure to clear things up so we can keep receiving
              pendingSocketTransactions.remove(socket);

              if (logger.level <= Logger.FINEST) {
                logger.log("bb[0].limit() " + bb[0].limit() + " bb[0].remaining() " + bb[0].remaining() + " from " + socket);
              }

              // deserialize the object
              SimpleInputBuffer sib = new SimpleInputBuffer(bb[0].array());

              short type = sib.readShort();

              PastMessage result = (PastMessage) endpoint.getDeserializer().deserialize(sib, type, (byte) 0, null);
              deliver(null, result);

            }

            // there will be more data on the socket if we haven't received everything yet
            // need to register either way to be able to read from the sockets when they are closed remotely, could alternatively close early
            // cause we are currently only sending 1 message per socket, but it's better to just keep reading in case we one day reuse sockets
            socket.register(true, false, 10000, this);

            // recursive call to handle next object
            // cant do this becasue calling read when not ready throws an exception
//          receiveSelectResult(socket, canRead, canWrite);
          } catch (IOException ioe) {
            receiveException(socket, ioe);
          }
        }

        public void receiveException(AppSocket socket, Exception e) {
          if (logger.level <= Logger.WARNING) {
            logger.logException("Error receiving message", e);
          }
          close(socket);
        }

        public void close(AppSocket socket) {
          if (socket == null) {
            return;
          }
          //System.out.println("Closing "+socket);
          pendingSocketTransactions.remove(socket);
          socket.close();
        }

      });
    endpoint.register();
  }

  /**
   * Gets the Environment attribute of the PastImpl object
   *
   * @return The Environment value
   */
  public Environment getEnvironment() {
    return environment;
  }

  /**
   * Returns of the outstanding messages. This is a DEBUGGING method ONLY!
   *
   * @return The list of all the outstanding messages
   */
  public Continuation[] getOutstandingMessages() {
    return (Continuation[]) outstanding.values().toArray(new Continuation[0]);
  }

  /**
   * Returns the endpoint associated with the Past - ONLY FOR TESTING - DO NOT
   * USE
   *
   * @return The endpoint
   */
  public Endpoint getEndpoint() {
    return endpoint;
  }

  /**
   * Returns a new uid for a message
   *
   * @return A new id
   */
  protected synchronized int getUID() {
    return id++;
  }

  /**
   * Returns a continuation which will respond to the given message.
   *
   * @param msg DESCRIBE THE PARAMETER
   * @return A new id
   */
  protected Continuation getResponseContinuation(final PastMessage msg) {
    if (logger.level <= Logger.FINER) {
      logger.log("Getting the Continuation to respond to the message " + msg);
    }
    final ContinuationMessage cmsg = (ContinuationMessage) msg;

    return
      new Continuation() {
        public void receiveResult(Object o) {
          cmsg.receiveResult(o);
          endpoint.route(null, cmsg, msg.getSource());
        }

        public void receiveException(Exception e) {
          cmsg.receiveException(e);
          endpoint.route(null, cmsg, msg.getSource());
        }
      };
  }

  /**
   * Do like above, but use a socket
   *
   * @param msg
   * @return
   */
  protected Continuation getFetchResponseContinuation(final PastMessage msg) {
    final ContinuationMessage cmsg = (ContinuationMessage) msg;

    return
      new Continuation() {
        public void receiveResult(Object o) {
          cmsg.receiveResult(o);
          PastContent content = (PastContent) o;
          if (socketStrategy.sendAlongSocket(SocketStrategy.TYPE_FETCH, content)) {
            sendViaSocket(msg.getSource(), cmsg, null);
          } else {
            endpoint.route(null, cmsg, msg.getSource());
          }
        }

        public void receiveException(Exception e) {
          cmsg.receiveException(e);
          endpoint.route(null, cmsg, msg.getSource());
        }
      };
  }

  /**
   * Internal method which returns the handles to an object. It first checks to
   * see if the handles can be determined locally, and if so, returns.
   * Otherwise, it sends a LookupHandles messsage out to find out the nodes.
   *
   * @param id The id to fetch the handles for
   * @param max The maximum number of handles to return
   * @param command The command to call with the result (NodeHandle[])
   */
  protected void getHandles(Id id, int max, Continuation command) {
    NodeHandleSet set = endpoint.replicaSet(id, max);

    if (set.size() == max) {
      command.receiveResult(set);
    } else {
      sendRequest(id, new LookupHandlesMessage(getUID(), id, max, getLocalNodeHandle(), id),
        new StandardContinuation(command) {
          public void receiveResult(Object o) {
            NodeHandleSet replicas = (NodeHandleSet) o;

            // check to make sure we've fetched the correct number of replicas
            if (endpoint.replicaSet(endpoint.getLocalNodeHandle().getId(), replicationFactor + 1).size() > replicas.size()) {
              parent.receiveException(new PastException("Only received " + replicas.size() + " replicas - cannot insert as we know about more nodes."));
            } else {
              parent.receiveResult(replicas);
            }
          }
        });
    }
  }

  /**
   * get the nodeHandle of the local Past node
   *
   * @return the nodehandle
   */
  public NodeHandle getLocalNodeHandle() {
    return endpoint.getLocalNodeHandle();
  }

  /**
   * Returns the number of replicas used in this Past
   *
   * @return the number of replicas for each object
   */
  public int getReplicationFactor() {
    return replicationFactor;
  }

  // ----- UTILITY METHODS -----

  /**
   * Returns the replica manager for this Past instance. Should *ONLY* be used
   * for testing. Messing with this will cause unknown behavior.
   *
   * @return This Past's replica manager
   */
  public Replication getReplication() {
    return replicaManager.getReplication();
  }

  /**
   * Returns this Past's storage manager. Should *ONLY* be used for testing.
   * Messing with this will cause unknown behavior.
   *
   * @return This Past's storage manager.
   */
  public StorageManager getStorageManager() {
    return storage;
  }

  /**
   * Gets the Instance attribute of the PastImpl object
   *
   * @return The Instance value
   */
  public String getInstance() {
    return instance;
  }

  /**
   * Sets the ContentDeserializer attribute of the PastImpl object
   *
   * @param deserializer The new ContentDeserializer value
   */
  public void setContentDeserializer(PastContentDeserializer deserializer) {
    contentDeserializer = deserializer;
  }

  /**
   * Sets the ContentHandleDeserializer attribute of the PastImpl object
   *
   * @param deserializer The new ContentHandleDeserializer value
   */
  public void setContentHandleDeserializer(PastContentHandleDeserializer deserializer) {
    contentHandleDeserializer = deserializer;
  }

  /**
   * DESCRIBE THE METHOD
   *
   * @return DESCRIBE THE RETURN VALUE
   */
  public String toString() {
    if (endpoint == null) {
      return super.toString();
    }
    return "PastImpl[" + endpoint.getInstance() + "]";
  }

  // ----- INTERNAL METHODS -----

  /**
   * Internal method which builds the replication manager. Can be overridden by
   * subclasses.
   *
   * @param node The node to base the RM off of
   * @param instance The instance name to use
   * @return The replication manager, ready for use
   */
  protected ReplicationManager buildReplicationManager(Node node, String instance) {
    return new ReplicationManagerImpl(node, this, replicationFactor, instance);
  }

  /**
   * DESCRIBE THE METHOD
   *
   * @param handle DESCRIBE THE PARAMETER
   * @param m DESCRIBE THE PARAMETER
   * @param c DESCRIBE THE PARAMETER
   */
  private void sendViaSocket(final NodeHandle handle, final PastMessage m, final Continuation c) {
    if (c != null) {
      CancellableTask timer = endpoint.scheduleMessage(new MessageLostMessage(m.getUID(), getLocalNodeHandle(), null, m, handle), MESSAGE_TIMEOUT);
      insertPending(m.getUID(), timer, c);
    }

    // create a bb[] to be written
    SimpleOutputBuffer sob = new SimpleOutputBuffer();
    try {
      sob.writeInt(0);
      // place holder for size...
      sob.writeShort(m.getType());
      m.serialize(sob);
    } catch (IOException ioe) {
      if (c != null) {
        c.receiveException(ioe);
      }
    }

    // add the size back to the beginning...
    int size = sob.getWritten() - 4;
    // remove the size of the size :)
    if (logger.level <= Logger.FINER) {
      logger.log("Sending size of " + size + " to " + handle + " to send " + m);
    }
    byte[] bytes = sob.getBytes();
    MathUtils.intToByteArray(size, bytes, 0);

    // prepare the bytes for writing
    final ByteBuffer[] bb = new ByteBuffer[1];
    bb[0] = ByteBuffer.wrap(bytes, 0, sob.getWritten());
    // the whole thing

    if (logger.level <= Logger.FINE) {
      logger.log("Opening socket to " + handle + " to send " + m);
    }
    endpoint.connect(handle,
      new AppSocketReceiver() {

        public void receiveSocket(AppSocket socket) {
          if (logger.level <= Logger.FINER) {
            logger.log("Opened socket to " + handle + ":" + socket + " to send " + m);
          }
          socket.register(false, true, 10000, this);
        }

        public void receiveSelectResult(AppSocket socket, boolean canRead,
                                        boolean canWrite) {
          if (logger.level <= Logger.FINEST) {
            logger.log("Writing to " + handle + ":" + socket + " to send " + m);
          }

          try {
//          ByteBuffer[] outs = new ByteBuffer[1];
//          ByteBuffer out = ByteBuffer.wrap(endpoint.getLocalNodeHandle().getId().toByteArray());
//          outs[0] = out;
//          socket.write(outs, 0, 1);

            socket.write(bb, 0, 1);
          } catch (IOException ioe) {
            if (c != null) {
              c.receiveException(ioe);
            } else
              if (logger.level <= Logger.WARNING) {
              logger.logException("Error sending " + m, ioe);
            }
            return;
            // don't continue to try to send
          }
          if (bb[0].remaining() > 0) {
            socket.register(false, true, 10000, this);
          } else {
            socket.close();
          }
        }

        public void receiveException(AppSocket socket, Exception e) {
          if (c != null) {
            c.receiveException(e);
          }
        }
      },
      10000);
  }


  /**
   * Sends a request message across the wire, and stores the appropriate
   * continuation.
   *
   * @param id The destination id
   * @param message The message to send.
   * @param command The command to run once a result is received
   */
  protected void sendRequest(Id id, PastMessage message, Continuation command) {
    sendRequest(id, message, null, command);
  }

  /**
   * Sends a request message across the wire, and stores the appropriate
   * continuation.
   *
   * @param handle The node handle to send directly too
   * @param message The message to send.
   * @param command The command to run once a result is received
   */
  protected void sendRequest(NodeHandle handle, PastMessage message, Continuation command) {
    sendRequest(null, message, handle, command);
  }

  /**
   * Sends a request message across the wire, and stores the appropriate
   * continuation. Sends the message using the provided handle as a hint.
   *
   * @param id The destination id
   * @param message The message to send.
   * @param command The command to run once a result is received
   * @param hint DESCRIBE THE PARAMETER
   */
  protected void sendRequest(Id id, PastMessage message, NodeHandle hint, Continuation command) {
    if (logger.level <= Logger.FINER) {
      logger.log("Sending request message " + message + " {" + message.getUID() + "} to id " + id + " via " + hint);
    }
    CancellableTask timer = endpoint.scheduleMessage(new MessageLostMessage(message.getUID(), getLocalNodeHandle(), id, message, hint), MESSAGE_TIMEOUT);
    insertPending(message.getUID(), timer, command);
    endpoint.route(id, message, hint);
  }

  /**
   * Loads the provided continuation into the pending table
   *
   * @param uid The id of the message
   * @param command The continuation to run
   * @param timer DESCRIBE THE PARAMETER
   */
  private void insertPending(int uid, CancellableTask timer, Continuation command) {
    if (logger.level <= Logger.FINER) {
      logger.log("Loading continuation " + uid + " into pending table");
    }
    timers.put(new Integer(uid), timer);
    outstanding.put(new Integer(uid), command);
  }

  /**
   * Removes and returns the provided continuation from the pending table
   *
   * @param uid The id of the message
   * @return The continuation to run
   */
  private Continuation removePending(int uid) {
    if (logger.level <= Logger.FINER) {
      logger.log("Removing and returning continuation " + uid + " from pending table");
    }
    CancellableTask timer = (CancellableTask) timers.remove(new Integer(uid));

    if (timer != null) {
      timer.cancel();
    }

    return (Continuation) outstanding.remove(new Integer(uid));
  }

  /**
   * Handles the response message from a request.
   *
   * @param message The message that arrived
   */
  private void handleResponse(PastMessage message) {
    if (logger.level <= Logger.FINE) {
      logger.log("handling reponse message " + message + " from the request");
    }
    Continuation command = removePending(message.getUID());

    if (command != null) {
      message.returnResponse(command, environment, instance);
    }
  }

  /**
   * Method which inserts the given object into the cache
   *
   * @param content The content to cache
   */
  private void cache(final PastContent content) {
    cache(content, new ListenerContinuation("Caching of " + content, environment));
  }

  /**
   * Method which inserts the given object into the cache
   *
   * @param content The content to cache
   * @param command The command to run once done
   */
  public void cache(final PastContent content, final Continuation command) {
    if (logger.level <= Logger.FINER) {
      logger.log("Inserting PastContent object " + content + " into cache");
    }

    if ((content != null) && (!content.isMutable())) {
      storage.cache(content.getId(), null, content, command);
    } else {
      command.receiveResult(new Boolean(true));
    }
  }

  /**
   * Internal method which actually performs an insert for a given object. Here
   * so that subclasses can override the types of insert messages which are sent
   * across the wire.
   *
   * @param builder The object which builds the messages
   * @param command The command to call once done
   * @param id DESCRIBE THE PARAMETER
   * @param useSocket DESCRIBE THE PARAMETER
   */
  protected void doInsert(final Id id, final MessageBuilder builder, Continuation command, final boolean useSocket) {
    // first, we get all of the replicas for this id
    getHandles(id, replicationFactor + 1,
      new StandardContinuation(command) {
        public void receiveResult(Object o) {
          NodeHandleSet replicas = (NodeHandleSet) o;
          if (logger.level <= Logger.FINER) {
            logger.log("Received replicas " + replicas + " for id " + id);
          }

          // then we send inserts to each replica and wait for at least
          // threshold * num to return successfully
          MultiContinuation multi =
            new MultiContinuation(parent, replicas.size()) {
              public boolean isDone() throws Exception {
                int numSuccess = 0;
                for (int i = 0; i < haveResult.length; i++) {
                  if ((haveResult[i]) && (Boolean.TRUE.equals(result[i]))) {
                    numSuccess++;
                  }
                }

                if (numSuccess >= (SUCCESSFUL_INSERT_THRESHOLD * haveResult.length)) {
                  return true;
                }

                if (super.isDone()) {
                  for (int i = 0; i < result.length; i++) {
                    if (result[i] instanceof Exception) {
                      if (logger.level <= Logger.WARNING) {
                        logger.logException("result[" + i + "]:", (Exception) result[i]);
                      }
                    }
                  }

                  throw new PastException("Had only " + numSuccess + " successful inserts out of " + result.length + " - aborting.");
                }
                return false;
              }

              public Object getResult() {
                Boolean[] b = new Boolean[result.length];
                for (int i = 0; i < b.length; i++) {
                  b[i] = new Boolean((result[i] == null) || Boolean.TRUE.equals(result[i]));
                }

                return b;
              }
            };

          for (int i = 0; i < replicas.size(); i++) {
            NodeHandle handle = replicas.getHandle(i);
            PastMessage m = builder.buildMessage();
            Continuation c = new NamedContinuation("InsertMessage to " + replicas.getHandle(i) + " for " + id, multi.getSubContinuation(i));
            if (useSocket) {
              sendViaSocket(handle, m, c);
            } else {
              sendRequest(handle, m, c);
            }
          }
        }
      });
  }


  // ----- PAST METHODS -----

  /**
   * Inserts an object with the given ID into this instance of Past.
   * Asynchronously returns a PastException to command, if the operation was
   * unsuccessful. If the operation was successful, a Boolean[] is returned
   * representing the responses from each of the replicas which inserted the
   * object.
   *
   * @param obj the object to be inserted
   * @param command Command to be performed when the result is received
   */
  public void insert(final PastContent obj, final Continuation command) {
    if (logger.level <= Logger.FINER) {
      logger.log("Inserting the object " + obj + " with the id " + obj.getId());
    }

    if (logger.level <= Logger.FINEST) {
      logger.log(" Inserting data of class " + obj.getClass().getName() + " under " + obj.getId().toStringFull());
    }

    doInsert(obj.getId(),
      new MessageBuilder() {
        public PastMessage buildMessage() {
          return new InsertMessage(getUID(), obj, getLocalNodeHandle(), obj.getId());
        }
      },
      new StandardContinuation(command) {
        public void receiveResult(final Object array) {
          cache(obj,
            new SimpleContinuation() {
              public void receiveResult(Object o) {
                parent.receiveResult(array);
              }
            });
        }
      },
      socketStrategy.sendAlongSocket(SocketStrategy.TYPE_INSERT, obj));
  }

  /**
   * Retrieves the object stored in this instance of Past with the given ID.
   * Asynchronously returns a PastContent object as the result to the provided
   * Continuation, or a PastException. This method is provided for convenience;
   * its effect is identical to a lookupHandles() and a subsequent fetch() to
   * the handle that is nearest in the network. The client must authenticate the
   * object. In case of failure, an alternate replica of the object can be
   * obtained via lookupHandles() and fetch(). This method is not safe if the
   * object is immutable and storage nodes are not trusted. In this case,
   * clients should used the lookUpHandles method to obtains the handles of all
   * primary replicas and determine which replica is fresh in an
   * application-specific manner.
   *
   * @param id the key to be queried
   * @param command Command to be performed when the result is received
   */
  public void lookup(final Id id, final Continuation command) {
    lookup(id, true, command);
  }

  /**
   * Method which performs the same as lookup(), but allows the callee to
   * specify if the data should be cached.
   *
   * @param id the key to be queried
   * @param cache Whether or not the data should be cached
   * @param command Command to be performed when the result is received
   */
  public void lookup(final Id id, final boolean cache, final Continuation command) {
    if (logger.level <= Logger.FINER) {
      logger.log(" Performing lookup on " + id.toStringFull());
    }

    storage.getObject(id,
      new StandardContinuation(command) {
        public void receiveResult(Object o) {
          if (o != null) {
            command.receiveResult(o);
          } else {
            // send the request across the wire, and see if the result is null or not
            sendRequest(id, new LookupMessage(getUID(), id, getLocalNodeHandle(), id),
              new NamedContinuation("LookupMessage for " + id, this) {
                public void receiveResult(final Object o) {
                  // if we have an object, we return it
                  // otherwise, we must check all replicas in order to make sure that
                  // the object doesn't exist anywhere
                  if (o != null) {
                    // lastly, try and cache object locally for future use
                    if (cache) {
                      cache((PastContent) o,
                        new SimpleContinuation() {
                          public void receiveResult(Object object) {
                            command.receiveResult(o);
                          }
                        });
                    } else {
                      command.receiveResult(o);
                    }
                  } else {
                    lookupHandles(id, replicationFactor + 1,
                      new Continuation() {
                        public void receiveResult(Object o) {
                          PastContentHandle[] handles = (PastContentHandle[]) o;

                          for (int i = 0; i < handles.length; i++) {
                            if (handles[i] != null) {
                              fetch(handles[i],
                                new StandardContinuation(parent) {
                                  public void receiveResult(final Object o) {
                                    // lastly, try and cache object locally for future use
                                    if (cache) {
                                      cache((PastContent) o,
                                        new SimpleContinuation() {
                                          public void receiveResult(Object object) {
                                            command.receiveResult(o);
                                          }
                                        });
                                    } else {
                                      command.receiveResult(o);
                                    }
                                  }
                                });

                              return;
                            }
                          }

                          // there were no replicas of the object
                          command.receiveResult(null);
                        }

                        public void receiveException(Exception e) {
                          command.receiveException(e);
                        }
                      });
                  }
                }

                public void receiveException(Exception e) {
                  // If the lookup message failed , we then try to fetch all of the handles, just
                  // in case.  This may fail too, but at least we tried.
                  receiveResult(null);
                }
              });
          }
        }
      });
  }

  /**
   * Retrieves the handles of up to max replicas of the object stored in this
   * instance of Past with the given ID. Asynchronously returns an array of
   * PastContentHandles as the result to the provided Continuation, or a
   * PastException. Each replica handle is obtained from a different primary
   * storage root for the the given key. If max exceeds the replication factor r
   * of this Past instance, only r replicas are returned. This method will
   * return a PastContentHandle[] array containing all of the handles.
   *
   * @param id the key to be queried
   * @param max the maximal number of replicas requested
   * @param command Command to be performed when the result is received
   */
  public void lookupHandles(final Id id, int max, final Continuation command) {
    if (logger.level <= Logger.FINE) {
      logger.log("Retrieving handles of up to " + max + " replicas of the object stored in Past with id " + id);
    }

    if (logger.level <= Logger.FINER) {
      logger.log("Fetching up to " + max + " handles of " + id.toStringFull());
    }

    getHandles(id, max,
      new StandardContinuation(command) {
        public void receiveResult(Object o) {
          NodeHandleSet replicas = (NodeHandleSet) o;
          if (logger.level <= Logger.FINER) {
            logger.log("Receiving replicas " + replicas + " for lookup Id " + id);
          }

          MultiContinuation multi =
            new MultiContinuation(parent, replicas.size()) {
              public Object getResult() {
                PastContentHandle[] p = new PastContentHandle[result.length];

                for (int i = 0; i < result.length; i++) {
                  if (result[i] instanceof PastContentHandle) {
                    p[i] = (PastContentHandle) result[i];
                  }
                }

                return p;
              }
            };

          for (int i = 0; i < replicas.size(); i++) {
            lookupHandle(id, replicas.getHandle(i), multi.getSubContinuation(i));
          }
        }
      });
  }

  /**
   * Retrieves the handle for the given object stored on the requested node.
   * Asynchronously returns a PostContentHandle (or null) to the provided
   * continuation.
   *
   * @param id the key to be queried
   * @param handle The node on which the handle is requested
   * @param command Command to be performed when the result is received
   */
  public void lookupHandle(Id id, NodeHandle handle, Continuation command) {
    if (logger.level <= Logger.FINE) {
      logger.log("Retrieving handle for id " + id + " from node " + handle);
    }

    sendRequest(handle, new FetchHandleMessage(getUID(), id, getLocalNodeHandle(), handle.getId()),
      new NamedContinuation("FetchHandleMessage to " + handle + " for " + id, command));
  }

  /**
   * Retrieves the object associated with a given content handle. Asynchronously
   * returns a PastContent object as the result to the provided Continuation, or
   * a PastException. The client must authenticate the object. In case of
   * failure, an alternate replica can be obtained using a different handle
   * obtained via lookupHandles().
   *
   * @param command Command to be performed when the result is received
   * @param handle DESCRIBE THE PARAMETER
   */
  public void fetch(PastContentHandle handle, Continuation command) {
    if (logger.level <= Logger.FINE) {
      logger.log("Retrieving object associated with content handle " + handle);
    }

    if (logger.level <= Logger.FINER) {
      logger.log("Fetching object under id " + handle.getId().toStringFull() + " on " + handle.getNodeHandle());
    }

    NodeHandle han = handle.getNodeHandle();
    sendRequest(han, new FetchMessage(getUID(), handle, getLocalNodeHandle(), han.getId()),
      new NamedContinuation("FetchMessage to " + handle.getNodeHandle() + " for " + handle.getId(), command));
  }


  // ----- COMMON API METHODS -----

  /**
   * This method is invoked on applications when the underlying node is about to
   * forward the given message with the provided target to the specified next
   * hop. Applications can change the contents of the message, specify a
   * different nextHop (through re-routing), or completely terminate the
   * message.
   *
   * @param message The message being sent, containing an internal message along
   *      with a destination key and nodeHandle next hop.
   * @return Whether or not to forward the message further
   */
  public boolean forward(final RouteMessage message) {
    Message internal;
    try {
      internal = message.getMessage(endpoint.getDeserializer());
    } catch (IOException ioe) {
      throw new RuntimeException(ioe);
    }

    if (internal instanceof LookupMessage) {
      final LookupMessage lmsg = (LookupMessage) internal;
      Id id = lmsg.getId();

      // if it is a request, look in the cache
      if (!lmsg.isResponse()) {
        if (logger.level <= Logger.FINER) {
          logger.log("Lookup message " + lmsg + " is a request; look in the cache");
        }
        if (storage.exists(id)) {
          // deliver the message, which will do what we want
          if (logger.level <= Logger.FINE) {
            logger.log("Request for " + id + " satisfied locally - responding");
          }
          deliver(endpoint.getId(), lmsg);
          return false;
        }
      }
    } else if (internal instanceof LookupHandlesMessage) {
      LookupHandlesMessage lmsg = (LookupHandlesMessage) internal;

      if (!lmsg.isResponse()) {
        if (endpoint.replicaSet(lmsg.getId(), lmsg.getMax()).size() == lmsg.getMax()) {
          if (logger.level <= Logger.FINE) {
            logger.log("Hijacking lookup handles request for " + lmsg.getId());
          }

          deliver(endpoint.getId(), lmsg);
          return false;
        }
      }
    }

    return true;
  }

  /**
   * This method is called on the application at the destination node for the
   * given id.
   *
   * @param id The destination id of the message
   * @param message The message being sent
   */
  public void deliver(Id id, Message message) {
    final PastMessage msg = (PastMessage) message;

    if (msg.isResponse()) {
      handleResponse((PastMessage) message);
    } else {
      if (logger.level <= Logger.INFO) {
        logger.log("Received message " + message + " with destination " + id);
      }

      if (msg instanceof InsertMessage) {
        final InsertMessage imsg = (InsertMessage) msg;

        // make sure the policy allows the insert
        if (policy.allowInsert(imsg.getContent())) {
          inserts++;
          final Id msgid = imsg.getContent().getId();

          lockManager.lock(msgid,
            new StandardContinuation(getResponseContinuation(msg)) {

              public void receiveResult(Object result) {
                storage.getObject(msgid,
                  new StandardContinuation(parent) {
                    public void receiveResult(Object o) {
                      try {
                        // allow the object to check the insert, and then insert the data
                        PastContent content = imsg.getContent().checkInsert(msgid, (PastContent) o);
                        storage.store(msgid, null, content,
                          new StandardContinuation(parent) {
                            public void receiveResult(Object result) {
                              getResponseContinuation(msg).receiveResult(result);
                              lockManager.unlock(msgid);
                            }
                          });
                      } catch (PastException e) {
                        parent.receiveException(e);
                      }
                    }
                  });
              }
            });
        } else {
          getResponseContinuation(msg).receiveResult(new Boolean(false));
        }
      } else if (msg instanceof LookupMessage) {
        final LookupMessage lmsg = (LookupMessage) msg;
        lookups++;

        // if the data is here, we send the reply, as well as push a cached copy
        // back to the previous node
        storage.getObject(lmsg.getId(),
          new StandardContinuation(getResponseContinuation(lmsg)) {
            public void receiveResult(Object o) {
              if (logger.level <= Logger.FINE) {
                logger.log("Received object " + o + " for id " + lmsg.getId());
              }

              // send result back
              parent.receiveResult(o);

              // if possible, pushed copy into previous hop cache
              if ((lmsg.getPreviousNodeHandle() != null) &&
                (o != null) &&
                (!((PastContent) o).isMutable())) {
                NodeHandle handle = lmsg.getPreviousNodeHandle();
                if (logger.level <= Logger.FINE) {
                  logger.log("Pushing cached copy of " + ((PastContent) o).getId() + " to " + handle);
                }

                CacheMessage cmsg = new CacheMessage(getUID(), (PastContent) o, getLocalNodeHandle(), handle.getId());
                //endpoint.route(null, cmsg, handle);
              }
            }
          });
      } else if (msg instanceof LookupHandlesMessage) {
        LookupHandlesMessage lmsg = (LookupHandlesMessage) msg;
        NodeHandleSet set = endpoint.replicaSet(lmsg.getId(), lmsg.getMax());
        if (logger.level <= Logger.FINER) {
          logger.log("Returning replica set " + set + " for lookup handles of id " + lmsg.getId() + " max " + lmsg.getMax() + " at " + endpoint.getId());
        }
        getResponseContinuation(msg).receiveResult(set);
      } else if (msg instanceof FetchMessage) {
        FetchMessage fmsg = (FetchMessage) msg;
        lookups++;

        Continuation c;
//        c = getResponseContinuation(msg);
        c = getFetchResponseContinuation(msg);
        // has to be special to determine how to send the message

        storage.getObject(fmsg.getHandle().getId(), c);
      } else if (msg instanceof FetchHandleMessage) {
        final FetchHandleMessage fmsg = (FetchHandleMessage) msg;
        fetchHandles++;

        storage.getObject(fmsg.getId(),
          new StandardContinuation(getResponseContinuation(msg)) {
            public void receiveResult(Object o) {
              PastContent content = (PastContent) o;

              if (content != null) {
                if (logger.level <= Logger.FINE) {
                  logger.log("Retrieved data for fetch handles of id " + fmsg.getId());
                }
                parent.receiveResult(content.getHandle(PastImpl.this));
              } else {
                parent.receiveResult(null);
              }
            }
          });
      } else if (msg instanceof CacheMessage) {
        cache(((CacheMessage) msg).getContent());
      } else {
        if (logger.level <= Logger.SEVERE) {
          logger.log("ERROR - Received message " + msg + "of unknown type.");
        }
      }
    }
  }

  /**
   * This method is invoked to inform the application that the given node has
   * either joined or left the neighbor set of the local node, as the set would
   * be returned by the neighborSet call.
   *
   * @param handle The handle that has joined/left
   * @param joined Whether the node has joined or left
   */
  public void update(NodeHandle handle, boolean joined) {
  }


  // ----- REPLICATION MANAGER METHODS -----

  /**
   * This upcall is invoked to tell the client to fetch the given id, and to
   * call the given command with the boolean result once the fetch is completed.
   * The client *MUST* call the command at some point in the future, as the
   * manager waits for the command to return before continuing.
   *
   * @param id The id to fetch
   * @param hint DESCRIBE THE PARAMETER
   * @param command DESCRIBE THE PARAMETER
   */
  public void fetch(final Id id, NodeHandle hint, Continuation command) {
    if (logger.level <= Logger.FINER) {
      logger.log("Sending out replication fetch request for the id " + id);
    }

    policy.fetch(id, hint, backup, this,
      new StandardContinuation(command) {
        public void receiveResult(Object o) {
          if (o == null) {
            if (logger.level <= Logger.WARNING) {
              logger.log("Could not fetch id " + id + " - policy returned null in namespace " + instance);
            }
            parent.receiveResult(new Boolean(false));
          } else {
            if (logger.level <= Logger.FINEST) {
              logger.log("inserting replica of id " + id);
            }

            if (!(o instanceof PastContent)) {
              if (logger.level <= Logger.WARNING) {
                logger.log("ERROR! Not PastContent " + o.getClass().getName() + " " + o);
              }
            }
            storage.getStorage().store(((PastContent) o).getId(), null, (PastContent) o, parent);
          }
        }
      });
  }

  /**
   * This upcall is to notify the client that the given id can be safely removed
   * from the storage. The client may choose to perform advanced behavior, such
   * as caching the object, or may simply delete it.
   *
   * @param id The id to remove
   * @param command DESCRIBE THE PARAMETER
   */
  public void remove(final Id id, Continuation command) {
    if (backup != null) {
      storage.getObject(id,
        new StandardContinuation(command) {
          public void receiveResult(Object o) {
            backup.cache(id, storage.getMetadata(id), (Serializable) o,
              new StandardContinuation(parent) {
                public void receiveResult(Object o) {
                  storage.unstore(id, parent);
                }
              });
          }
        });
    } else {
      storage.unstore(id, command);
    }
  }

  /**
   * This upcall should return the set of keys that the application currently
   * stores in this range. Should return a empty IdSet (not null), in the case
   * that no keys belong to this range.
   *
   * @param range the requested range
   * @return DESCRIBE THE RETURN VALUE
   */
  public IdSet scan(IdRange range) {
    return storage.getStorage().scan(range);
  }

  /**
   * This upcall should return the set of keys that the application currently
   * stores. Should return a empty IdSet (not null), in the case that no keys
   * belong to this range.
   *
   * @return DESCRIBE THE RETURN VALUE
   */
  public IdSet scan() {
    return storage.getStorage().scan();
  }

  /**
   * This upcall should return whether or not the given id is currently stored
   * by the client.
   *
   * @param id The id in question
   * @return Whether or not the id exists
   */
  public boolean exists(Id id) {
    return storage.getStorage().exists(id);
  }

  /**
   * DESCRIBE THE METHOD
   *
   * @param id DESCRIBE THE PARAMETER
   * @param command DESCRIBE THE PARAMETER
   */
  public void existsInOverlay(Id id, Continuation command) {
    lookupHandles(id, replicationFactor + 1,
      new StandardContinuation(command) {
        public void receiveResult(Object result) {
          Object results[] = (Object[]) result;
          for (int i = 0; i < results.length; i++) {
            if (results[i] instanceof PastContentHandle) {
              parent.receiveResult(Boolean.TRUE);
              return;
            }
          }
          parent.receiveResult(Boolean.FALSE);
        }
      });
  }

  /**
   * DESCRIBE THE METHOD
   *
   * @param id DESCRIBE THE PARAMETER
   * @param command DESCRIBE THE PARAMETER
   */
  public void reInsert(Id id, Continuation command) {
    storage.getObject(id,
      new StandardContinuation(command) {
        public void receiveResult(final Object o) {
          insert((PastContent) o,
            new StandardContinuation(parent) {
              public void receiveResult(Object result) {
                Boolean results[] = (Boolean[]) result;
                for (int i = 0; i < results.length; i++) {
                  if (results[i].booleanValue()) {
                    parent.receiveResult(Boolean.TRUE);
                    return;
                  }
                }
                parent.receiveResult(Boolean.FALSE);
              }
            });
        }
      });
  }


  /**
   * DESCRIBE THE CLASS
   *
   * @version $Id: pretty.settings 2305 2005-03-11 20:22:33Z jeffh $
   * @author jeffh
   */
  protected class PastDeserializer implements MessageDeserializer {
    /**
     * DESCRIBE THE METHOD
     *
     * @param buf DESCRIBE THE PARAMETER
     * @param type DESCRIBE THE PARAMETER
     * @param priority DESCRIBE THE PARAMETER
     * @param sender DESCRIBE THE PARAMETER
     * @return DESCRIBE THE RETURN VALUE
     * @exception IOException DESCRIBE THE EXCEPTION
     */
    public Message deserialize(InputBuffer buf, short type, byte priority,
                               NodeHandle sender) throws IOException {
      try {
        switch (type) {
          case CacheMessage.TYPE:
            return CacheMessage.build(buf, endpoint, contentDeserializer);
          case FetchHandleMessage.TYPE:
            return FetchHandleMessage.build(buf, endpoint, contentHandleDeserializer);
          case FetchMessage.TYPE:
            return FetchMessage.build(buf, endpoint, contentDeserializer, contentHandleDeserializer);
          case InsertMessage.TYPE:
            return InsertMessage.build(buf, endpoint, contentDeserializer);
          case LookupHandlesMessage.TYPE:
            return LookupHandlesMessage.build(buf, endpoint);
          case LookupMessage.TYPE:
            return LookupMessage.build(buf, endpoint, contentDeserializer);
        }
      } catch (IOException e) {
        if (logger.level <= Logger.SEVERE) {
          logger.log("Exception in deserializer in " + PastImpl.this.endpoint.toString() + ":" + instance + " " + e);
        }
        throw e;
      }
      throw new IllegalArgumentException("Unknown type:" + type + " in " + PastImpl.this.toString());
    }
  }

  /**
   * Class which builds a message
   *
   * @version $Id: pretty.settings 2305 2005-03-11 20:22:33Z jeffh $
   * @author jeffh
   */
  public interface MessageBuilder {
    /**
     * DESCRIBE THE METHOD
     *
     * @return DESCRIBE THE RETURN VALUE
     */
    public PastMessage buildMessage();
  }

}
