}

    /**
     * Write a error message to the log facility.
     *
     * @param msg the message
     */
    private void error(String msg)
    {
        logger.error(toString() + ": " + msg);
    }

    /**
     * Handle a alternate packet.
     *
     * @param pkt the packet
     */
    private void handle(AlternatePacket pkt)
    {
        IAgent agent = getAgent(pkt.getStream());

        if (agent != null)
        {
            agent.setAlternate(pkt.getLayer(), pkt.getSender(),
                pkt.getAltLayer(), pkt.getAlternates());
        }
    }

    /**
     * Handle a data packet.
     *
     * @param pkt the packet
     */
    private void handle(DataPacket pkt)
    {
        debug("data from " + pkt.getSender() + " using stream "
            + pkt.getStream());

        IMulticastAgent multicast = getMulticast(pkt.getStream());

        if (multicast != null)
        {
            multicast.data(pkt.getVSender(), pkt.getVDest(),
                pkt.getForwardType(), pkt.getSource(), pkt.getSeqNr(),
                pkt.getMask(), pkt.getData(), pkt.getSendTimestamp());
        }
    }

    /**
     * Handle a elect new leader packet.
     *
     * @param pkt the packet
     */
    private void handle(ElectNewLeaderPacket pkt)
    {
        IAgent agent = getAgent(pkt.getStream());

        if (agent != null)
        {
            agent.electNewLeader(pkt.getSender());
        }
    }

    /**
     * Handle a forward request packet.
     *
     * @param pkt the packet
     */
    private void handle(ForwardRequestPacket pkt)
    {
        IMulticastAgent multicast = getMulticast(pkt.getStream());

        if (multicast != null)
        {
            multicast.forwardRequest(pkt.getVSender(), pkt.getVDest(),
                pkt.getForwardType(), pkt.getSource(), pkt.getSeqNrs());
        }
    }

    /**
     * Handle a heartbeat packet.
     *
     * @param pkt the packet
     */
    private void handle(HeartbeatPacket pkt)
    {
        debug("heartbeat from " + pkt.getSender() + " using stream "
            + pkt.getStream());

        IAgent agent = getAgent(pkt.getStream());

        if (agent != null)
        {
            agent.heartbeat(pkt.getLayer(), pkt.getSender(), pkt.getSuccessors());
        }
    }

    /**
     * Handle a join packet.
     *
     * @param pkt the packet
     */
    private void handle(JoinPacket pkt)
    {
        IAgent agent = getAgent(pkt.getStream());

        if (agent != null)
        {
            agent.join(pkt.getLayer(), pkt.getSender());
        }
    }

    /**
     * Handle a join redirect packet.
     *
     * @param pkt the packet
     */
    private void handle(JoinRedirectPacket pkt)
    {
        IAgent agent = getAgent(pkt.getStream());

        if (agent != null)
        {
            agent.joinRedirect(pkt.getLayer(), pkt.getSender(),
                pkt.getRedirect());
        }
    }

    /**
     * Handle a leave packet.
     *
     * @param pkt the packet
     */
    private void handle(LeavePacket pkt)
    {
        IAgent agent = getAgent(pkt.getStream());

        if (agent != null)
        {
            agent.leave(pkt.getSender());
        }
    }

    /**
     * Handle a neighbor packet.
     *
     * @param pkt the packet
     */
    private void handle(NeighborPacket pkt)
    {
        IAgent agent = getAgent(pkt.getStream());

        if (agent != null)
        {
            agent.setNeighbors(pkt.getLayer(), pkt.getSender(),
                pkt.getNeighbors());
        }
    }

    /**
     * Handle a new leader packet.
     *
     * @param pkt the packet
     */
    private void handle(NewLeaderPacket pkt)
    {
        IAgent agent = getAgent(pkt.getStream());

        if (agent != null)
        {
            agent.newLeader(pkt.getLayer(), pkt.getSender(), pkt.getNewLeader());
        }
    }

    /**
     * Handle a overlay information packet.
     * @param pkt the packet
     */
    private void handle(OverlayInfoPacket pkt)
    {
        IAgent agent = getAgent(pkt.getStream());

        if (agent != null)
        {
            agent.overlayInfo(pkt.getSender(), pkt.getOverlayInfo());
        }
    }

    /**
     * Handle a query packet.
     *
     * @param pkt the packet
     */
    private void handle(QueryPacket pkt)
    {
        IAgent agent = getAgent(pkt.getStream());

        if (agent != null)
        {
            agent.query(pkt.getLayer(), pkt.getSender());
        }
    }

    /**
     * Handle a query response packet.
     *
     * @param pkt the packet
     */
    private void handle(QueryResponsePacket pkt)
    {
        IAgent agent = getAgent(pkt.getStream());

        if (agent != null)
        {
            agent.queryResponse(pkt.getLayer(), pkt.getSender(),
                pkt.getQueryResponse());
        }
    }

    /**
     * Handle a query rp response packet.
     *
     * @param pkt the packet
     */
    private void handle(QueryRpResponsePacket pkt)
    {
        IAgent agent = getAgent(pkt.getStream());

        if (agent != null)
        {
            agent.queryRpResponse(pkt.getQueryResponse(), pkt.getMagic());
        }
    }

    /**
     * Handle a super cluster packet.
     *
     * @param pkt the packet
     */
    private void handle(SuperClusterPacket pkt)
    {
        IAgent agent = getAgent(pkt.getStream());

        if (agent != null)
        {
            agent.setSuperCluster(pkt.getLayer(), pkt.getSender(),
                pkt.getLeader(), pkt.getAlternates(), pkt.getSuperCluster());
        }
    }

    /**
     * Handle a transfer leader packet.
     *
     * @param pkt the packet
     */
    private void handle(TransferLeaderPacket pkt)
    {
        IAgent agent = getAgent(pkt.getStream());

        if (agent != null)
        {
            agent.transferLeader(pkt.getLayer(), pkt.getSender(),
                pkt.getNewLeader(), pkt.getSuperLeader(), pkt.getMembers(),
                pkt.isMerge());
        }
    }

        touch();

        return (long) ((info.getAvg() + (4.0 * info.getStd())) / 2.0);
    }

    /**
     * Get the experienced loss rate to the given agent. The difference between
     * the loss rate and the cost depends on the application. In some cases it
     * may be the same, in others it differs. The important thing about the
     * loss rate is that its unit is in %, whereas the cost could be any unit.
     *
     * @return the loss rate to the agent, percentage
     */
    public double getLossRate()
    {
        SocketAddress sa = agentId.getSocketAddress();

        NetworkInfo networkInfo = sock.getNetworkManagerInstanceUnreliable()
                                      .get(sa);
        double lossRate = networkInfo.getLossRate();
        QueueInfo queueRxInfo = sock.getInstanceRx().get(null);
        double dropRateRx = queueRxInfo.getDropRateAvg();
        QueueInfo queueTxInfo = sock.getInstanceTx().get(sa);
        double dropRateTx = queueTxInfo.getDropRateAvg();

        double res = (1.0 - lossRate);

        res *= (1.0 - dropRateRx);
        res *= (1.0 - dropRateTx);
        res = 1.0 - res;

        return Math.min(1.0, Math.max(0.0, res));
    }

    /**
     * Get the last access time.
     *
     * @return the last access time
     */
    public ITimestamp getLastAccess()
    {
        return lastAccess;
    }

    /**
     * Get the agent id.
     *
     * @return the agent id
     */
    public AgentId getAgentId()
    {
        return agentId;
    }

    /**
     * Creates a human readable string representation of the object.
     *
     * @return the string representation
     */
    public String toString()
    {
        return "cost info for " + agentId;

            handle((LeavePacket) packet);
        }
        else if (packet instanceof NewLeaderPacket)
        {
            handle((NewLeaderPacket) packet);
        }
        else if (packet instanceof AlternatePacket)
        {
            handle((AlternatePacket) packet);
        }
        else if (packet instanceof ElectNewLeaderPacket)
        {
            handle((ElectNewLeaderPacket) packet);
        }
        else if (packet instanceof NeighborPacket)
        {
            handle((NeighborPacket) packet);
        }
        else if (packet instanceof QueryPacket)
        {
            handle((QueryPacket) packet);
        }
        else if (packet instanceof QueryResponsePacket)
        {
            handle((QueryResponsePacket) packet);
        }
        else if (packet instanceof QueryRpResponsePacket)
        {
            handle((QueryRpResponsePacket) packet);
        }
        else if (packet instanceof SuperClusterPacket)
        {
            handle((SuperClusterPacket) packet);
        }
        else if (packet instanceof TransferLeaderPacket)

    public void setup(IPacketSocketFactory socketFactory,
        ITimestampFactory tsmpFactory,
        IEpochServiceFactory epochServiceFactory, IReefConfiguration config,
        SocketAddress[] bootstrap)
    {
        this.config = (Configuration) config;

        if ((nodeId == null)
            || !nodeId.getSocketAddress().equals(socketFactory
                .getLocalSocketAddress()))
        {
            nodeId = AgentId.createAgentId((InetSocketAddress) socketFactory
                    .getLocalSocketAddress(), tsmpFactory);
        }

        sock = socketFactory.register(ENemoPacketType.NAMESPACE,
                this.config.getUseLowPriority());
        sock.getTypeMapping().registerAll(ReefTypeMapping.class);
        sock.getPacketMapping().registerAll(NemoPacketMapping.class);

        epochService = epochServiceFactory.create(this.config);

        // create the callback for the agent
        callback = new Callback(sock);

                    BandwidthCostInfo info = (BandwidthCostInfo) e.getValue();

                    if (info.getLastAccess().before(deadline))
                    {
                        iter.remove();
                    }
                }
            }
        }
    }

    /**
     * Probe all the super cluster members.
     */
    private void probe()
    {
        for (int i = 0; i < agents.length; i++)
        {
            IAgent agent = agents[i];
            AgentId[] superCluster = agent.getSuperCluster();
            Packet packet = new LatencyProbePacket();

            for (int j = 0; j < superCluster.length; j++)
            {
                packet.setSender(agent.getId());
                packet.setSocketAddress(superCluster[j].getSocketAddress());

                try
                {
                    sock.send(SendQueue.DEFAULT_PRIORITY, new Message(packet),

    {
        IPacket packet = msg.getPacket();

        if (packet instanceof DataPacket)
        {
            handle((DataPacket) packet);
        }
        else if (packet instanceof ForwardRequestPacket)
        {
            handle((ForwardRequestPacket) packet);
        }
        else if (packet instanceof HeartbeatPacket)
        {
            handle((HeartbeatPacket) packet);
        }
        else if (packet instanceof JoinPacket)
        {
            handle((JoinPacket) packet);
        }
        else if (packet instanceof JoinRedirectPacket)
        {
            handle((JoinRedirectPacket) packet);
        }
        else if (packet instanceof LeavePacket)
        {
            handle((LeavePacket) packet);
        }
        else if (packet instanceof NewLeaderPacket)

                res = new BandwidthCostInfo(agent, sock, tsmpFactory);
                instances.put(agent, res);
            }
        }

        return res;
    }

    /**
     * The epoch implementation.
     */
    public void epoch()
    {
        if (UNIFORM.nextDoubleFromTo(0.0, 1.0) < ESTIMATE_PROB)
        {
            probe();
        }

        if (UNIFORM.nextDoubleFromTo(0.0, 1.0) < CLEANUP_PROB)
        {
            synchronized (instances)
            {
                Iterator iter = instances.entrySet().iterator();
                ITimestamp deadline = tsmpFactory.create(-MAX_STORAGE);

                while (iter.hasNext())
                {
                    Map.Entry e = (Map.Entry) iter.next();