context: /*
  This program is part of BruNet, a library for the creation of efficient overlay
  networks.
  Copyright (C) 2005  University of California
  
  This program 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 2
  of the License, or (at your option) any later version.
  
  This program 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 this program; if not, write to the Free Software
  Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
  */
  
  using System;
  using System.Collections;
  using System.Collections.Generic;
  using System.Collections.Specialized;
  using System.Threading;
  
  using Brunet.Collections;
  using Brunet.Connections;
  using BCon = Brunet.Concurrent;
  using Brunet.Util;
  using Brunet.Transport;
  using MR = Brunet.Services.MapReduce;
  using Brunet.Messaging;
  
  using Brunet.Concurrent;
  using Brunet.Symphony;
  namespace Brunet
  {
  
    /**
     * This class represents endpoint of packet communication.
     * There may be many nodes on each computer on the Brunet
     * Network.  An example may be :  a user wants to chat, and
     * share a file.  The file may be represented as a Node on
     * the Brunet Network, and also the chat user may represent
     * itself as a Node on the network.  Both of these Nodes
     * reside on the same computer host.
     *
     * The Node also keeps itself connected and manages its
     * connections. 
     * 
     */
    abstract public class Node : IDataHandler, ISender, IActionQueue
    {
      static Node() {
        SenderFactory.Register("localnode", CreateLocalSender); 
      }
  
      public static Node CreateLocalSender(object n, string uri) {
        return (Node)n;
      }
      /**
       * Create a node in the realm "global"
       */
      protected Node(Address addr) : this(addr, "global") { }
      /**
       * Create a node with a given local address and
       * a set of Routers.
       * @param addr Address for the local node
       * @param realm the Realm or Namespace this node belongs to
       */
  
      protected Node(Address addr, string realm)
      {
        //Start with the address hashcode:
  
        _sync = new Object();
        lock(_sync)
        {
          DemuxHandler = new DemuxHandler();
          /*
           * Make all the hashtables : 
           */
          _local_add = AddressParser.Parse( addr.ToMemBlock() );
          _realm = String.Intern(realm);
          
          /* Set up the heartbeat */
          _heart_period = 500; //500 ms, or 1/2 second.
          _heartbeat_handlers = new Dictionary<EventHandler, Brunet.Util.FuzzyEvent>();
  
          _task_queue = new NodeTaskQueue(this);
          _packet_queue = new BCon.LFBlockingQueue<IAction>();
  
          _running = 0;
          _send_pings = 1;
          _LOG = ProtocolLog.Monitor.Enabled;
  
          _connection_table = new ConnectionTable(_local_add);
          _connection_table.ConnectionEvent += this.ConnectionHandler;
  
          //We start off offline.
          _con_state = Node.ConnectionState.Offline;
          
          /* Set up the ReqrepManager as a filter */
          _rrm = new ReqrepManager(this);
          DemuxHandler.GetTypeSource(PType.Protocol.ReqRep).Subscribe(_rrm, null);
          _rrm.Subscribe(this, null);
          this.HeartBeatEvent += _rrm.TimeoutChecker;
          /* Set up RPC */
          _rpc = new RpcManager(_rrm);
          DemuxHandler.GetTypeSource( PType.Protocol.Rpc ).Subscribe(_rpc, null);
  
          /*
           * Where there is a change in the Connections, we might have a state
           * change
           */
          _connection_table.ConnectionEvent += this.CheckForStateChange;
          _connection_table.DisconnectionEvent += this.CheckForStateChange;
          _connection_table.StatusChangedEvent += this.CheckForStateChange;
  
          _codeinjection = new Brunet.Services.CodeInjection(this);
          _codeinjection.LoadLocalModules();
          /*
           * We must later make sure the EdgeEvent events from
           * any EdgeListeners are connected to _cph.EdgeHandler
           */
          /**
           * Here are the protocols that every edge must support
           */
          /* Here are the transport addresses */
          _remote_ta = new ArrayList();
          /*@throw ArgumentNullException if the list ( new ArrayList()) is null.
           */
          /* EdgeListener's */
          _edgelistener_list = new ArrayList();
          _edge_factory = new EdgeFactory();
          
          /* Initialize this at 15 seconds */
          _connection_timeout = new TimeSpan(0,0,0,0,15000);
          //Check the edges from time to time
          IAction cec_act = new HeartBeatAction(this, this.CheckEdgesCallback);
          _check_edges = Brunet.Util.FuzzyTimer.Instance.DoEvery(delegate(DateTime dt) {
            this.EnqueueAction(cec_act);
          }, 15000, 1000);
        }
      }
   //////////////
   ///  Inner Classes
   //////////
   
      protected class HeartBeatAction : IAction {
      
        readonly Node _n;
        readonly EventHandler _eh;
      
        public HeartBeatAction(Node n, EventHandler eh) {
          _n = n;
          _eh = eh;
        }
        public void Start() {
          try {
            _eh(_n, EventArgs.Empty);
          }
          catch(Exception x) {
            ProtocolLog.WriteIf(ProtocolLog.Exceptions, String.Format(
              "Exception in heartbeat event : {0}", x));
          }
        }
     }
  
      protected class LogAction : IAction {
        protected readonly BCon.LFBlockingQueue<IAction> _q;
        public LogAction(BCon.LFBlockingQueue<IAction> q) {
          _q = q;
        }
  
        public void Start() {
          ProtocolLog.Write(
          ProtocolLog.Monitor, String.Format("I am alive: {0}, packet_queue_length: {1}", 
                                                              DateTime.UtcNow, _q.Count));
        }
      }
  
      /**
       * When we do announces using the seperate thread, this is
       * what we pass
       */
      private class AnnounceState : IAction {
        public readonly MemBlock Data;
        public readonly ISender From;
        public readonly Node LocalNode;
        public AnnounceState(Node n, MemBlock p, ISender from) {
          LocalNode = n;
          Data = p;
          From = from;
        }
        
        /**
         * Perform the action of announing a packet
         */
        public void Start() {
          LocalNode.Announce(Data, From);
        }
        public override string ToString() {
          try {
            return Data.GetString(System.Text.Encoding.ASCII);
          }
          catch {
            return "AnnounceState: could not get string as ASCII";
          }
        }
      }
      private class EdgeCloseAction : IAction {
        protected Edge EdgeToClose;
        public EdgeCloseAction(Edge e) {
          EdgeToClose = e;
        }
        public void Start() {
          EdgeToClose.Close();
        }
        public override string ToString() {
          return "EdgeCloseAction: " + EdgeToClose.ToString();
        }
      }
  
      public class GracefulCloseAction : Triple<Node, Edge, string>, IAction {
        public GracefulCloseAction(Node n, Edge e, string r) : base(n, e, r) { }
        public void Start() {
          First.GracefullyClose(Second, Third);
        }
      }
  
      /**
       * This is a TaskQueue where new TaskWorkers are started
       * by EnqueueAction, so they are executed in the announce thread
       * and without the call stack growing arbitrarily
       */
      protected class NodeTaskQueue : TaskQueue {
        protected readonly Node LocalNode;
        public NodeTaskQueue(Node n) {
          LocalNode = n;
        }
        protected override void Start(TaskWorker tw) {
          LocalNode.EnqueueAction(tw);
        }
      }
  
  //////
  // End of inner classes
  /////
  
      public delegate bool EdgeVerifier(Node node, Edge e, Address addr);
      public EdgeVerifier EdgeVerifyMethod;
  
      /**
       * This represents the Connection state of the node.
       * We use different words for each state to reduce the
       * liklihood of typos causing problems in the code.
       */
      public enum ConnectionState {
        Offline, /// Not yet called Node.Connect
        Joining, /// Called Node.Connect but IsConnected has never been true
        Connected, /// IsConnected is true.
        SeekingConnections, /// We were previously Connected, but lost connections.
        Leaving, /// Node.Disconnect has been called, but we haven't closed all edges.
        Disconnected /// We are completely disconnected and have no active Edges.
      }
      public delegate void StateChangeHandler(Node n, ConnectionState newstate);
      /**
       * This event is called every time Node.ConState changes.  The new state
       * is passed with the event.
       */
      public event StateChangeHandler StateChangeEvent;
      public ConnectionState ConState {
        get {
          lock( _sync ) {
            return _con_state;
          }
        }
      }
      protected ConnectionState _con_state;
  
      protected readonly Address _local_add;
      /**
       * The Address of this Node
       */
      public Address Address
      {
        get
        {
          return _local_add;
        }
      }
      protected readonly EdgeFactory _edge_factory;
      /**
       *  my EdgeFactory
       */
      public EdgeFactory EdgeFactory { get { return _edge_factory; } }
  
      /**
       * Here are all the EdgeListener objects for this Node
       */
      protected ArrayList _edgelistener_list;
      public ArrayList EdgeListenerList {
        get {
          return (ArrayList) _edgelistener_list.Clone();
        }
      }
  
      /**
       * These are all the local TransportAddress objects that
       * refer to EdgeListener objects attached to this node.
       * This IList is ReadOnly
       */
      public IList LocalTAs {
        get {
          var local_ta = new List<TransportAddress>();
          var enums = new List<IEnumerable>();
          foreach(EdgeListener el in _edgelistener_list) {
            enums.Add( el.LocalTAs );
          }
          //Go round robin through all the LocalTAs:
          var uncast_tas = new Functional.Interleave(enums);
          //Cast the resulting IEnumerable into IEnumerable<TransportAddress>
          var all_tas = new Functional.CastEnumerable<TransportAddress>(uncast_tas);
          //Make sure we don't keep too many of these things:
          local_ta.AddRange(new Functional.Take<TransportAddress>(all_tas, _MAX_RECORDED_TAS));
          return local_ta;
        }
      }
  
      /**
       * This is an estimate of the current
       * network size.  It is not an exact
       * value.
       *
       * A value of -1 means there is not
       * enough information to make a meaningful
       * estimate.
       */
      virtual public int NetworkSize {
        get { return -1; }
      }
      protected readonly BCon.LFBlockingQueue<IAction> _packet_queue;
      /** The IAction that was most recently started */
      protected IAction _current_action;
      protected float _packet_queue_exp_avg = 0.0f;
  
      //Getting from a BooleanSwitch is strangely very expensive, do it once
      protected bool _LOG;
      //If we get this big, we just throw an exception, and not enqueue it
      protected static readonly int MAX_QUEUE_LENGTH = 8192;
      /**
       * This number should be more thoroughly tested, but my system and dht
       * never surpassed 105.
       */
      public static readonly int MAX_AVG_QUEUE_LENGTH = 4096;
      public static readonly float PACKET_QUEUE_RETAIN = 0.99f;
      public bool DisconnectOnOverload {
        get { return _disconnect_on_overload; }
        set { _disconnect_on_overload = value; }
      }
  
      public bool _disconnect_on_overload = false;
  
      protected readonly string _realm;
      /**
       * Each Brunet Node is in exactly 1 realm.  This is 
       * a namespacing feature.  This allows you to create
       * Brunets which are separate from other Brunets.
       *
       * The default is "global" which is the standard
       * namespace.
       */
      public string Realm { get { return _realm; } }
      
      protected readonly ArrayList _remote_ta;
      /**
       * These are all the remote TransportAddress objects that
       * this Node may use to connect to remote Nodes
       *
       * This can be shared between nodes or not.
       *
       * This is the ONLY proper way to set the RemoteTAs for this
       * node.
       */
      public ArrayList RemoteTAs {
        get {
          return _remote_ta;
        }
        set {
          UpdateRemoteTAs(value);
        }
      }
  
      /**
       * This is true after Connect is called and false after
       * Disconnect is called.
       */
      protected int _running;
      protected int _send_pings;
      protected Util.FuzzyEvent _check_edges;
  
      /** Object which we lock for thread safety */
      protected readonly object _sync;
  
      /**  <summary>Handles subscriptions for the different types of packets
      that come to this node.</summary>*/
      public readonly DemuxHandler DemuxHandler;
  
      /**
      <summary>All packets that come to this are demultiplexed according to t.
      To subscribe or unsubscribe, get the ISource for the type you want and
      subscribe to it,</summary>
      <param name="t">The key for the MultiSource.</param>
      @deprecated Use Node.DemuxHandler.GetTypeSource
      */
      public ISource GetTypeSource(Object t) {
        return DemuxHandler.GetTypeSource(t);
      }
  
      /**
      <summary>Deletes (and thus unsubscribes) all IDataHandlers for a given key.
      </summary>
      <param name="t">The key for the MultiSource.</param>
      @deprecated Use Node.DemuxHandler.ClearTypeSource
      */
      public void ClearTypeSource(Object t) {
        DemuxHandler.ClearTypeSource(t);
      }
  
      /**
      <summary>Deletes (and thus unsubscribes) all IDataHandlers for the table.
      </summary>
      @deprecated Use Node.DemuxHandler.Clear
      */
      public void Clear() {
        DemuxHandler.Clear();
      }
  
      protected readonly ConnectionTable _connection_table;
  
      /**
       * Manages the various mappings associated with connections
       */
      public virtual ConnectionTable ConnectionTable { get { return _connection_table; } }
      /**
       * Brunet IPHandler service!
       */
      public IPHandler IPHandler { get { return _iphandler; } }
      protected IPHandler _iphandler;
      protected Brunet.Services.CodeInjection _codeinjection;
      
      protected readonly ReqrepManager _rrm;
      public ReqrepManager Rrm { get { return _rrm; } }
      protected readonly RpcManager _rpc;
      public RpcManager Rpc { get { return _rpc; } }
      protected MR.MapReduceHandler _mr_handler;
      public MR.MapReduceHandler MapReduce { get { return _mr_handler; } }
  
  
      /**
       * This is true if the Node is properly connected in the network.
       * If you want to know when it is safe to assume you are connected,
       * listen to all for Node.ConnectionTable.ConnectionEvent and
       * Node.ConnectionTable.DisconnectionEvent and then check
       * this property.  If it is true, you should probably wait
       * until it is false if you need the Node to be connected
       */
      public abstract bool IsConnected { get; }
      protected readonly NodeTaskQueue _task_queue;
      /**
       * This is the TaskQueue for this Node
       */
      public TaskQueue TaskQueue { get { return _task_queue; } }
  
      protected int _heart_period;
      ///how many milliseconds between heartbeats
      public int HeartPeriod { get { return _heart_period; } }
  
      ///If we don't hear anything from a *CONNECTION* in this time, ping it.
      protected TimeSpan _connection_timeout;
      ///This is the maximum value we allow _connection_timeout to grow to
      protected static readonly TimeSpan MAX_CONNECTION_TIMEOUT = new TimeSpan(0,0,0,15,0);
      //Give edges this long to get connected, then drop them
      protected static readonly TimeSpan _unconnected_timeout = new TimeSpan(0,0,0,30,0);
      /**
       * Maximum number of TAs we keep in both for local and remote.
       * This does not control how many we send to our neighbors.
       */
      static protected readonly int _MAX_RECORDED_TAS = 10000;
  
      ///after each HeartPeriod, the HeartBeat event is fired
      public event EventHandler HeartBeatEvent {
        add {
          IAction hba = new HeartBeatAction(this, value);
          Action<DateTime> torun = delegate(DateTime now) {
            //Execute the code in the node's thread
            this.EnqueueAction(hba);
          };
          //every period +/- half a period, run this event
          var fe = Brunet.Util.FuzzyTimer.Instance.DoEvery(torun, _heart_period, _heart_period / 2 + 1);
          lock( _sync ) {
            _heartbeat_handlers[ value ] = fe;
          }
        }
  
        remove {
          Brunet.Util.FuzzyEvent fe = null;
          lock( _sync ) {
            if(_heartbeat_handlers.TryGetValue(value, out fe) ) {
              _heartbeat_handlers.Remove(value);
            }
          }
          if( fe != null ) {
            fe.TryCancel();
          }
        }
      }
  
      protected Dictionary<EventHandler, Brunet.Util.FuzzyEvent> _heartbeat_handlers;
      
      //add an event handler which conveys the fact that Disconnect has been called on the node
      public event EventHandler DepartureEvent;
  
      //add an event handler which conveys the fact that Connect has been called on the node
      public event EventHandler ArrivalEvent;
  
      public virtual void AddEdgeListener(EdgeListener el)
      {
        /* The EdgeFactory needs to be made aware of all EdgeListeners */
        _edge_factory.AddListener(el);
        _edgelistener_list.Add(el);
        /**
         * It is ESSENTIAL that the EdgeEvent of EdgeListener objects
         * be connected to the EdgeHandler method of ConnectionPacketHandler
         */
        el.EdgeEvent += this.EdgeHandler;
        el.EdgeCloseRequestEvent += delegate(object elsender, EventArgs args) {
          EdgeCloseRequestArgs ecra = (EdgeCloseRequestArgs)args;
          Close(ecra.Edge);
        };
      }
  
      /** Immediately stop the node. ONLY USE FOR TESTING!!!!!!!
       * This simulates a node being disconnected immediately from the network,
       * as in a crash, or a network outage.  It should NEVER be used by a
       * well-behaved node.  It is only for test programs.
       *
       * To gracefully disconnect, you the Node.Disconnect() method, which
       * closes edges gracefully and informs its neighbors before going offline.
       */
      public abstract void Abort();
      
      /**
       * Called when there is a connection or disconnection.  Send a StateChange
       * event if need be.
       * We could be transitioning from:
       *   Joining -> Connected
       *   Connected -> SeekingConnections
       */
      protected void CheckForStateChange(object ct, EventArgs ce_args) {
        bool con = this.IsConnected; 
        ConnectionState new_state;
        if( con ) {
          new_state = Node.ConnectionState.Connected;
        }
        else {
          /*
           * The only other state change that is triggered by a Connection
           * or Disconnection event is SeekingConnections
           */
          new_state = Node.ConnectionState.SeekingConnections;
        }
        bool success;
        SetConState(new_state, out success);
        if( success ) {
          SendStateChange(new_state);
        }
      }
  
      protected void Close(Edge e) {
        try {
          //This can throw an exception if the _packet_queue is closed
          EnqueueAction(new EdgeCloseAction(e));
        }
        catch {
          e.Close();
        }
      }
  
      /**
       * This Handler should be connected to incoming EdgeEvent
       * events.  If it is not, it cannot hear the new edges.
       *
       * When a new edge is created, we make sure we can hear
       * the packets from it.  Also, we make sure we can hear
       * the CloseEvent.
       *
       * @param edge the new Edge
       */
      protected void EdgeHandler(object edge, EventArgs args)
      {
        Edge e = (Edge)edge;
        try {
          _connection_table.AddUnconnected(e);
          e.Subscribe(this, e);
        }
        catch(TableClosedException) {
          /*
           * Close this edge immediately, before any packets
           * have a chance to be received.  We are shutting down,
           * and it is best that we stop getting new packets
           */
          e.Close();
        }
      }
  
      /**
       * Put this IAction object into the announce thread and call start on it
       * there.
       */
      public void EnqueueAction(IAction a) {
  #if BRUNET_SIMULATOR
        a.Start();
        return;
  #else
        int queue_size = _packet_queue.Count;
        if( queue_size > MAX_QUEUE_LENGTH ) {
          /*
           * Disconnect actually assumes the _packet_queue is being processed
           * if it is not, due to a blocking operation, or deadlock, we could
           * still be adding things into the queue.  This is here to prevent
           * a memory explosion
           */
  //        throw new Exception(String.Format("Queue is too long: {0}", queue_size));
          ProtocolLog.WriteIf(ProtocolLog.Exceptions, String.Format("Queue is too long: {0}", queue_size));
        }
        int count = _packet_queue.Enqueue(a);
        _packet_queue_exp_avg = (PACKET_QUEUE_RETAIN * _packet_queue_exp_avg)
            + ((1 - PACKET_QUEUE_RETAIN) * count);
  
        if(_packet_queue_exp_avg > MAX_AVG_QUEUE_LENGTH) {
          if(_LOG) {
            String top_string = String.Empty;
            try {
              top_string = _current_action.ToString();
            }
            catch {}
            ProtocolLog.Write(ProtocolLog.Monitor, String.Format(
              "Packet Queue Average too high: {0} at {1}.  Actual length:  {2}\n\tTop most action: {3}",
              _packet_queue_exp_avg, DateTime.UtcNow, count, top_string));
          }
          if(_disconnect_on_overload) {
            Disconnect();
          }
        }
  #endif
      }
  
      /**
       * Send the StateChange event
       */
      protected void SendStateChange(ConnectionState new_state) {
        if( new_state == Node.ConnectionState.Joining && ArrivalEvent != null) {
          ArrivalEvent(this, null);
        }
        if( new_state == Node.ConnectionState.Leaving && DepartureEvent != null) {
          DepartureEvent(this, null);
        }
        StateChangeEvent(this, new_state);
      }
      /**
       * This sets the ConState to new_cs and returns the old
       * ConState.
       *
       * This method knows about the allowable state transitions.
       * @param success is set to false if we can't do the state transition.
       * @return the value of ConState prior to the method being called
       */
      protected ConnectionState SetConState(ConnectionState new_cs, out bool success) {
        ConnectionState old_state;
        success = false;
        lock( _sync ) {
          old_state = _con_state;
          if( old_state == new_cs ) {
            //This is not a state change
            return old_state;
          }
          if( new_cs == Node.ConnectionState.Joining ) {
            success = (old_state == Node.ConnectionState.Offline);
          }
          else if( new_cs == Node.ConnectionState.Connected ) {
            success = (old_state == Node.ConnectionState.Joining) ||
                      (old_state == Node.ConnectionState.SeekingConnections);
          }
          else if( new_cs == Node.ConnectionState.SeekingConnections ) {
            success = (old_state == Node.ConnectionState.Connected);
          }
          else if( new_cs == Node.ConnectionState.Leaving ) {
            success = (old_state != Node.ConnectionState.Disconnected);
          }
          else if( new_cs == Node.ConnectionState.Disconnected ) {
            success = (old_state == Node.ConnectionState.Leaving );
          }
          else if( new_cs == Node.ConnectionState.Offline ) {
            // We can never move into the Offline state.
            success = false;
          }
          /*
           * Now let's update _con_state
           */
  
gt:         if( success ) {