context: /*
  Copyright (C) 2009  David Wolinsky <davidiw@ufl.edu>, University of Florida
  Copyright (C) 2007  David Wolinsky <davidiw@ufl.edu>, University of Florida
  
  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.Diagnostics;
  using System.Text;
  using System.Collections;
  using System.Security.Cryptography;
  using System.Threading;
  
  using Brunet;
  
  /**
  \namespace Brunet::DistributedServices 
  \brief Provides Distributed data storage services using the Brunet P2P
  infrastructure.
  */
  namespace Brunet.DistributedServices {
    /**
    <summary>Exception generated from the Dht should use this exception</summary>
    */
    public class DhtException: Exception {
      public DhtException(string message): base(message) {}
    }
  
    public class DhtPutException: DhtException {
      public readonly int Total;
      public readonly int Positive;
      public readonly int Negative;
  
      public DhtPutException(int total, int positive, int negative) :
        base(string.Format("Operation failed Total/Positive/Negative: {0}/{1}/{2}", total, positive, negative))
      {
        Total = total;
        Negative = negative;
        Positive = positive;
      }
    }
  
    /// <summary>This class provides a client interface to the dht, the servers only
    /// work together on a neighboring basis but not on a whole system basis.  It is
    /// up to the client to provide fault tolerance.  This class does it by naive
    /// replication.</summary>
    /// <remarks>This class implements the fault tolerant portion of the Dht by using
    /// naive replication as well as fixing holes in the dht.  A hole occurs when
    /// enough data results are received during a get to confirm existence, but not
    /// all the results returned a value.  In the case of a hole, a put will be used
    /// to place the data back to seal the hole.</remarks>
    public class Dht : IDht {
      /// <summary>The log enabler for the dht.</summary>
      public static BooleanSwitch DhtLog = new BooleanSwitch("Dht", "Log for Dht!");
      /// <summary>Lock for the dht put/get state tables.</summary>
      protected object _sync;
      /// <summary>The RpcManager to perform transactions through.</summary>
      protected RpcManager _rpc;
      /// <summary>The node to provide services for.</summary>
      public readonly Node Node;
      /// <summary>How many replications are made.</summary>
      public readonly int DEGREE;
      /// <summary>How long to wait for synchronous results.</summary>
      public readonly int DELAY;
      /// <summary>floor(DEGREE/2) + 1, the amount of positive results for a
      /// successful operation</summary>
      public readonly int MAJORITY;
      /// <summary>The state table for asynchronous puts.</summary>
      protected volatile Hashtable _adps_table = new Hashtable();
      /// <summary>The state table for asynchronos gets.</summary>
      protected volatile Hashtable _adgs_table = new Hashtable();
      /// <summary>True unless node is in offline, leaving, or disconnected</summary>
      protected bool _online;
      public bool Online { get { return _online; } }
      public string Name { get { return Node.Address.ToString(); } }
  
      /// <summary>A default Dht client provides a DEGREE of 1 and a sychronous wait
      /// time of up to 60 seconds.</summary>
      /// <param name ="node">The node to provide service for.</param>
      public Dht(Node node) {
        Node = node;
        _rpc = node.Rpc;
        _sync = new object();
        StateChangeHandler(Node, Node.ConState);
        Node.StateChangeEvent += StateChangeHandler;
        DEGREE = 1;
        MAJORITY = 1;
        DELAY = 60000;
      }
  
      /// <summary>Allows the user to specify the power of two of degrees to use.
      /// That is if degree=n, DEGREE for the dht is 2^n.</summary>
      /// <param name="node">The node to provide service for.</param>
      /// <param name="degree">n where DEGREE=2^n amount of replications to perform.
      /// </param>
      public Dht(Node node, int degree) : this(node) {
        this.DEGREE = (int) System.Math.Pow(2, degree);
        this.MAJORITY = (DEGREE / 2) + 1;
      }
  
      /// <summary>Allows the user to specify the power of two of degrees to use.
      /// That is if degree=n, DEGREE for the dht is 2^n.</summary>
      /// <param name="node">The node to provide service for.</param>
      /// <param name="degree">n where DEGREE=2^n amount of replications to
      /// perform.</param>
      /// <param name="delay">User specified delay for synchronous calls in
      /// seconds.</param>
      public Dht(Node node, int degree, int delay) : this(node, degree) {
        DELAY = delay * 1000;
      }
  
      /// <summary>We're online once we are seeking connections, we're offline
      /// once Disconnect has been called.</summary>
      protected void StateChangeHandler(Node n, Node.ConnectionState state) {
        lock(_sync) {
          if(state == Node.ConnectionState.Leaving ||
              state == Node.ConnectionState.Disconnected ||
              state == Node.ConnectionState.Offline) {
            _online = false;
          } else {
            _online = true;
          }
        }
      }
  
  
      public void AsyncCreate(MemBlock key, MemBlock value, int ttl, Channel returns) {
        AsyncPut(key, value, ttl, returns, true);
      }
  
      public bool Create(MemBlock key, MemBlock value, int ttl) {
        return Put(key, value, ttl, true);
      }
  
      public Hashtable[] Get(MemBlock key) {
        BlockingQueue returns = new BlockingQueue();
        AsyncGet(key, returns);
        ArrayList values = new ArrayList();
        while(true) {
          // Still a chance for Dequeue to execute on an empty closed queue 
          // so we'll do this instead.
          try {
            Hashtable hash = (Hashtable) returns.Dequeue();
            values.Add(hash);
          }
          catch (Exception) {
            break;
          }
        }
        return (Hashtable[]) values.ToArray(typeof(Hashtable));
      }
  
      /// <remarks>This starts the get process by sending dht.Get to all the remote
      /// end points that contain the key we're looking up.  The next step is
      /// is when the results are placed in the channel and GetEnqueueHandler is
      /// called or GetCloseHandler is called.  This means the get needs to be
      /// stateful, that information is stored in the _adgs_table.</remarks>
      public void AsyncGet(MemBlock key, Channel returns) {
        if(!_online) {
          throw new DhtException("The Node is (going) offline, DHT is offline.");
        }
        // create a GetState and map in our table map its queues to it
        // so when we get a GetHandler we know which state to load
        AsDhtGetState adgs = new AsDhtGetState(returns);
        Channel[] q = new Channel[DEGREE];
        lock(_adgs_table.SyncRoot) {
          for (int k = 0; k < DEGREE; k++) {
            Channel queue = new Channel(1);
            _adgs_table[queue] = adgs;
            q[k] = queue;
          }
        }
  
        // Setting up our Channels
        for (int k = 0; k < DEGREE; k++) {
          Channel queue = q[k];
          queue.EnqueueEvent += this.GetEnqueueHandler;
          queue.CloseEvent += this.GetCloseHandler;
          adgs.queueMapping[queue] = k;
        }
  
        // Sending off the request!
        adgs.brunet_address_for_key = MapToRing(key);
        for (int k = 0; k < DEGREE; k++) {
          Address target = new AHAddress(adgs.brunet_address_for_key[k]);
          AHSender s = new AHGreedySender(Node, target);
          // 1024 is in there for backwards compatibility
          _rpc.Invoke(s, q[k], "dht.Get", adgs.brunet_address_for_key[k], 1024, null);
        }
      }
  
      /// <summary>This is called as a result of a successful retrieval of data
      /// from a remote end point and performs follow up gets for remaining values
      /// </summary>
      /// <remarks>This adds the results to the entry in the _adgs_table.  Once a
      /// value has been received by a majority of nodes, it is enqueued into the
      /// requestors returns channel.  If not all results were retrieved follow up
      /// gets are performed, this is determined by looking at the state of the
      /// token, a non-null token implies there are remaining results.</remarks>
      /// <param name="o">The channel used to store the results.</param>
      /// <param name="args">Unused.</param>
      public void GetEnqueueHandler(Object o, EventArgs args) {
        Channel queue = (Channel) o;
        // Looking up state
        AsDhtGetState adgs = (AsDhtGetState) _adgs_table[queue];
  
        if(adgs == null) {
          return;
        }
  
        int idx = (int) adgs.queueMapping[queue];
        // Test to see if we got any results and place them into results if necessary
        ISender sendto = null;
        MemBlock token = null;
        try {
          RpcResult rpc_reply = (RpcResult) queue.Dequeue();
          ArrayList result = (ArrayList) rpc_reply.Result;
          //Result may be corrupted
          if(result == null) {
            throw new Exception("Invalid result");
          }
          ArrayList values = (ArrayList) result[0];
          int remaining = (int) result[1];
          if(remaining > 0) {
            token = MemBlock.Reference((byte[]) result[2]);
            sendto = rpc_reply.ResultSender;
          }
  
          // Going through the return values and adding them to our
          // results, if a majority of our servers say a data exists
          // we say it is a valid data and return it to the caller
          foreach (Hashtable ht in values) {
            MemBlock mbVal = MemBlock.Reference((byte[]) ht["value"]);
            int count = 1;
            Hashtable res = null;
            lock(adgs.SyncRoot) {
              res = (Hashtable) adgs.results[mbVal];
              if(res == null) {
                res = new Hashtable();
                adgs.results[mbVal] = res;
                adgs.ttls[mbVal] = ht["ttl"];
              }
              else {
                adgs.ttls[mbVal] = (int) adgs.ttls[mbVal] + (int) ht["ttl"];
              }
  
              res[idx] = true;
              count = ((ICollection) adgs.results[mbVal]).Count;
            }
            if(count == MAJORITY) {
              ht["ttl"] = (int) adgs.ttls[mbVal] / MAJORITY;
              adgs.returns.Enqueue(ht);
            }
          }
        }
        catch (Exception) {
          sendto = null;
          token = null;
        }
  
      // We were notified that more results were available!  Let's go get them!
        if(token != null && sendto != null) {
          Channel new_queue = new Channel(1);
          lock(adgs.SyncRoot) {
            adgs.queueMapping[new_queue] = idx;
          }
          lock(_adgs_table.SyncRoot) {
            _adgs_table[new_queue] = adgs;
          }
          new_queue.EnqueueEvent += this.GetEnqueueHandler;
          new_queue.CloseEvent += this.GetCloseHandler;
          try {
            _rpc.Invoke(sendto, new_queue, "dht.Get", 
                      adgs.brunet_address_for_key[idx], token);
          }
          catch(Exception) {
            lock(adgs.SyncRoot) {
              adgs.queueMapping.Remove(new_queue);
            }
            lock(_adgs_table.SyncRoot) {
              _adgs_table.Remove(new_queue);
            }
            new_queue.EnqueueEvent -= this.GetEnqueueHandler;
            new_queue.CloseEvent -= this.GetCloseHandler;
          }
        }
      }
  
      /// <summary>This is called by the Get callbacks when all the results for a
      /// get have come in.  This looks at the results, finds holes, and does a
      /// follow up put to place the data back into the dht via GetFollowUp.
      /// </summary>
      /// <param name="o">The channel representing a specific get.</param>
      /// <param name="args">Unused.</param>
      protected void GetCloseHandler(object o, EventArgs args) {
        Channel queue = (Channel) o;
        queue.EnqueueEvent -= this.GetEnqueueHandler;
        queue.CloseEvent -= this.GetCloseHandler;
        // Looking up state
        AsDhtGetState adgs = (AsDhtGetState) _adgs_table[queue];
  
  
gt:       if(adgs == null) {