# Router.java

    /*
    ** Module   : Router.java
    ** Abstract : the routing part of the protocol
    **
    ** Copyright (c) 2005-2023, Golden Code Development Corporation.
    **
    ** -#- -I- --Date-- --JPRM-- ----------------Description-----------------
    ** 001 SIY 20050121   @19505 Created initial version.
    ** 002 SIY 20050207   @19709 Added handling of the address resolution
    **                           requests.
    ** 003 SIY 20050315   @20389 Organized imports and other minor cleanups.
    ** 004 SIY 20050329   @20548 Fixed comments.
    ** 005 NVS 20050418   @20764 New Notifiable required dummy initEIR().
    ** 006 GES 20070111   @31794 Code cleanup and added documentation.
    ** 007 GES 20070115   @31837 Added support for conversation mode.
    ** 008 NVS 20070411   @32931 Router threads are created as part of a
    **                           thread group named "router".
    ** 009 ECF 20071101   @35887 Refactored net package. Added firewall stubs.
    **                           Enabled router-to-router connections and
    **                           forwarding. Integrated SessionManager.
    **                           Implemented generics. Other minor cleanup.
    ** 010 SVL 20090715   @43195 Added OOME protection into routing processing
    **                           thread.
    ** 011 GES 20090722   @43336 Better thread names to enhance debugging.
    ** 012 GES 20090816   @43637 Harden the route() thread to keep it running even
    **                           if exceptions/errors occur (previously it would
    **                           just exit and once all router threads are dead,
    **                           then no logins can finish). Fixed an address leak
    **                           when the queue stopped before INIT_CONVERSATION
    **                           could complete. That same race condition now has
    **                           additional protection.
    ** 013 GBB 20230512          Logging methods replaced by CentralLogger/ConversionStatus.
    */
    /*
    ** This program is free software: you can redistribute it and/or modify
    ** it under the terms of the GNU Affero General Public License as
    ** published by the Free Software Foundation, either version 3 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 Affero General Public License for more details.
    **
    ** You may find a copy of the GNU Affero GPL version 3 at the following
    ** location: https://www.gnu.org/licenses/agpl-3.0.en.html
    **
    ** Additional terms under GNU Affero GPL version 3 section 7:
    **
    **   Under Section 7 of the GNU Affero GPL version 3, the following additional
    **   terms apply to the works covered under the License.  These additional terms
    **   are non-permissive additional terms allowed under Section 7 of the GNU
    **   Affero GPL version 3 and may not be removed by you.
    **
    **   0. Attribution Requirement.
    **
    **     You must preserve all legal notices or author attributions in the covered
    **     work or Appropriate Legal Notices displayed by works containing the covered
    **     work.  You may not remove from the covered work any author or developer
    **     credit already included within the covered work.
    **
    **   1. No License To Use Trademarks.
    **
    **     This license does not grant any license or rights to use the trademarks
    **     Golden Code, FWD, any Golden Code or FWD logo, or any other trademarks
    **     of Golden Code Development Corporation. You are not authorized to use the
    **     name Golden Code, FWD, or the names of any author or contributor, for
    **     publicity purposes without written authorization.
    **
    **   2. No Misrepresentation of Affiliation.
    **
    **     You may not represent yourself as Golden Code Development Corporation or FWD.
    **
    **     You may not represent yourself for publicity purposes as associated with
    **     Golden Code Development Corporation, FWD, or any author or contributor to
    **     the covered work, without written authorization.
    **
    **   3. No Misrepresentation of Source or Origin.
    **
    **     You may not represent the covered work as solely your work.  All modified
    **     versions of the covered work must be marked in a reasonable way to make it
    **     clear that the modified work is not originating from Golden Code Development
    **     Corporation or FWD.  All modified versions must contain the notices of
    **     attribution required in this license.
    */

    package com.goldencode.p2j.net;

    import java.util.*;
    import java.util.logging.*;
    import com.goldencode.p2j.util.*;
    import com.goldencode.p2j.util.logging.*;

    /**
     * This singleton class is responsible for delivering messages between nodes
     * which have no direct connection.  It is also responsible for processing
     * <code>INIT_*</code> and <code>ADDRESS_REQUEST</code> messages.  These
     * messages are used for dynamic address assignment by leaf nodes and
     * for routing information lookup respectively.
     * <p>
     * Firewall capability is partially implemented.  Hooks are available to
     * filter inbound, outbound, and forwarded messages.  At the time of writing,
     * no filtering is active, so all messages are allowed to pass through these
     * hooks.
     */
    final class Router
    implements MessageTypes
    {
       /** Logger. */
       private static final CentralLogger LOG =
          CentralLogger.get(Router.class.getName());

       /** thread group for the router threads */
       private static ThreadGroup routeThreads = new ThreadGroup(
          Thread.currentThread().getThreadGroup(), "router");

       /** Routing message queue (shared instance). */
       private static final ThreadSafeQueue<InboundRequest> routingQueue =
          new ThreadSafeQueue<InboundRequest>();

       /** Singleton instance of this class. */
       private static Router router = null;

       /** The maximum valid address which could belong to this routing node. */
       private int boundary;

       /** Last allocated address. */
       private int lastAddress;

       /** Local node address. */
       private int nodeAddress;

       /** Routing table. */
       private HashMap<Integer, Queue> routingTable;

       /** Local nodes registry. */
       private HashMap<Integer, Queue> leafTable;

       /** Message processing thread pool. */
       private Thread[] threads;

       /**
        * Constructs an instance of node address using specified node address and
        * specified number of threads in the thread pool.
        *
        * @param    address
        *           The local node's address address.
        * @param    num
        *           Number of threads in the thread pool.
        */
       private Router(int address, int num)
       {
          if (num <= 0)
             num = 2;

          nodeAddress = address;
          lastAddress = address;

          boundary = nodeAddress | AddressServer.NODE_MASK;

          threads = new Thread[num];

          routingTable = new HashMap<Integer, Queue>();
          leafTable    = new HashMap<Integer, Queue>();

          // init thread pool
          for (int i = 0; i < threads.length; i++)
          {
             threads[i] = new Thread(routeThreads, new RouterStub());
             threads[i].setName(String.format("Router %d", i));
             threads[i].setDaemon(true);
             threads[i].start();
          }
       }

       /**
        * Return the singleton instance of this class if it is already running
        * or create a new instance if not. Note that parameters are only honored
        * when this method is called the first time, otherwise the parameters
        * are ignored.
        *
        * @param    address
        *           The local node's address address.
        * @param    num
        *           Number of threads in the thread pool.
        */
       static synchronized void initRouter(int address, int num)
       {
          if (router == null)
          {
             router = new Router(address, num);
          }
       }

       /**
        * Forward an inbound request which is not destined for this network node.
        *
        * @param   request
        *          Request to be forwarded.
        * @throws  InterruptedException
        *          when the queue is terminated.
        */
       static void forward(InboundRequest request)
       throws InterruptedException
       {
          routingQueue.enqueue(request);
       }

       /**
        * Terminate static queue.
        */
       static void terminateQueue()
       {
          routingQueue.terminate();
       }

       /**
        * This method is called when a queue has been disabled, but before it has
        * been stopped.  Clean up the router's routing and leaf tables.
        *
        * @param    queue
        *           The queue that is stopping.
        */
       static void queueStopping(Queue queue)
       {
          router.cleanUpQueue(queue);
       }

       /**
        * Filter an inbound message according to installed firewall rules (if any)
        * for inbound messages.
        * <p>
        * The current implementation is a stub only, always returning
        * <code>true</code>.
        *
        * @param   message
        *          Message to be tested.
        *
        * @return  <code>true</code> if the message should be processed,
        *          <code>false</code> if it should be dropped.
        */
       static boolean filterInbound(Message message)
       {
          return true;
       }

       /**
        * Filter an outbound message according to installed firewall rules (if
        * any) for outbound messages.
        * <p>
        * The current implementation is a stub only, always returning
        * <code>true</code>.
        *
        * @param   message
        *          Message to be tested.
        *
        * @return  <code>true</code> if the message should be processed,
        *          <code>false</code> if it should be dropped.
        */
       static boolean filterOutbound(Message message)
       {
          return true;
       }

       /**
        * Filter a forwarded message according to installed firewall rules (if
        * any) for forwarded messages.
        * <p>
        * The current implementation is a stub only, always returning
        * <code>true</code>.
        *
        * @param   message
        *          Message to be tested.
        *
        * @return  <code>true</code> if the message should be processed,
        *          <code>false</code> if it should be dropped.
        */
       static boolean filterForwarded(Message message)
       {
          return true;
       }

       /**
        * Main message processing procedure. It retrieves messages from routing
        * queue and processes them.  The is the core processing for each thread
        * in the router's thread pool.
        */
       private void route()
       {
          while (true)
          {
             InboundRequest request = null;

             try
             {
                // dequeue the next message (this will block and it may throw
                // interrupted exception)
                request = routingQueue.dequeue();

                // get the data out
                Message message = request.getMessage();

                // firewall/message filtering
                if (!filterForwarded(message))
                {
                   continue;
                }

                int destAddr = message.getKey().getDestinationAddress();

                // forward message not destined for this node;  doesn't support
                // hops, only directly connected nodes
                // NOTE:  destination address will be 0 for INIT_* message types,
                // since an address must be assigned, so we do not forward these
                if (destAddr != 0 && destAddr != nodeAddress)
                {
                   Queue destQueue = routingTable.get(destAddr);

                   if (destQueue != null)
                   {
                      destQueue.enqueueOutbound(message);
                   }

                   continue;
                }

                Queue  queue   = request.getQueue();
                Object payload = message.getPayload();
                int    type    = message.getType();

                // a leaf node is establishing a new session with this node and
                // it is registering an address that it has defined
                if (type == INIT_STANDARD || type == INIT_CONVERSATION)
                {
                   boolean   assigned = false;
                   Exception problem  = null;

                   try
                   {
                      if (!(payload instanceof AddressPair))
                      {
                         String err = "Invalid payload";
                         message.setPayload(new ProtocolViolation(err));
                      }
                      else
                      {
                         AddressPair pair = (AddressPair) payload;
                         int addr = pair.getNodeAddress();

                         // did the remote node provide an address?
                         if (addr == 0)
                         {
                            SessionManager sessMgr = SessionManager.get();

                            // create the provider side of the direct session being
                            // established;  this must happen before the address is
                            // generated and before the conversation is enabled,
                            // since both need to access the session; if the session
                            // creation was aborted, discard the message and iterate
                            if (!sessMgr.createDirectSession(queue))
                               continue;

                            // no address was specified so we must generate the
                            // next valid one
                            generateAddress(queue);
                            assigned = true;

                            // implement deferred/negotiated switch into
                            // conversation mode and start a daemon thread to drive
                            // it
                            if (type == INIT_CONVERSATION)
                            {
                               queue.enableConversation(true, true);
                            }

                            // the queue has both local and remote addresses assigned
                            // so return these addresses to the other side
                            message.setPayload(queue.getAddressPair());
                         }
                         else
                         {
                            if ((addr & AddressServer.NODE_MASK) == 0)
                            {
                               // router node
                               String err = "Invalid router initialization";
                               message.setPayload(new ProtocolViolation(err));
                            }
                            /*
                            else
                            {
                               // leaf node
                               allocateAddress(queue, addr);
                               message.setPayload(queue.getAddressPair());
                            }
                            */
                         }
                      }

                      message.setType(INIT_REPLY);
                      queue.enqueueOutbound(message);

                      queue.sessionInitialized();

                      // done with leaf session init, iterate for next message
                      continue;
                   }

                   catch (Exception exc)
                   {
                      problem = exc;
                   }

                   finally
                   {
                      // exceptions are not expected unless the client/socket
                      // disconnects before initialization is complete
                      if (queue.isRunning())
                      {
                         if (problem != null)
                         {
                            // unexpected issue
                            throw problem;
                         }
                      }
                      else
                      {
                         // special case, the race condition (between session exit
                         // and full session initialization) was triggered
                         if (assigned)
                         {
                            // normally this is processed by the queue stop logic,
                            // but in this race condition the order is reversed
                            cleanUpQueue(queue);
                         }
                      }
                   }
                }

                if (type == INIT_ROUTER)
                {
                   if (!(payload instanceof AddressPair))
                   {
                      String err = "Invalid payload";
                      message.setPayload(new ProtocolViolation(err));
                   }
                   else
                   {
                      AddressPair pair = (AddressPair) payload;
                      int addr = pair.getNodeAddress();

                      // did the remote node provide an address?
                      if (addr != 0 && (addr & AddressServer.NODE_MASK) == 0)
                      {
                         // router node
                         addRoutingNode(queue, addr);

                         // the queue has both local and remote addresses assigned
                         // so return these addresses to the other side
                         message.setPayload(queue.getAddressPair());
                      }
                      else
                      {
                         String err =
                            "Invalid node address for router initialization";
                         message.setPayload(new ProtocolViolation(err));
                      }
                   }

                   message.setType(INIT_REPLY);
                   queue.enqueueOutbound(message);
                   continue;
                }

                if (type == ADDRESS_REQUEST)
                {
                   // TODO: implement this! the previous implementation was not
                   //       useful, the "fill in some of the NodeInfo fields and
                   //       then send it" approach is being removed
                   // WARNING: it is likely that this message type itself will
                   //          be changed or become part of a larger sub-protocol
                   //          for registering/deregistering routing info and for
                   //          lookups
                   message.setPayload(null);
                   message.setType(ADDRESS_REPLY);
                   queue.enqueueOutbound(message);
                   continue;
                }
             }

             catch (Throwable thr)
             {
                try
                {
                   // deal with the server being shutdown
                   if (thr instanceof InterruptedException)
                   {
                      if (routingQueue.isTerminated())
                      {
                         // no more routing to do because the server is stopping
                         break;
                      }
                   }

                   // other than shutdown, we ignore errors or exceptions that
                   // occur during this loop because to exit would destabilize
                   // the server

                   // log that there was an unexpected failure
                   if (LOG.isLoggable(Level.SEVERE))
                   {
                      LOG.logp(Level.SEVERE,
                               "Router.route()",
                               "",
                               "failure in routing loop",
                               thr);
                   }

                   // release the client (otherwise it will hang)
                   if (request != null)
                   {
                      Exception reason = (thr instanceof Error) ? new Exception(thr)
                                                                : (Exception) thr;

                      request.getQueue().stop(reason, false);
                   }

                   if (thr instanceof OutOfMemoryError)
                   {
                      // delay to try to recover from the OOME
                      try
                      {
                         Thread.sleep(5000);
                      }
                      catch (InterruptedException e)
                      {
                         // ignore
                      }
                   }
                }

                catch (Throwable t)
                {
                   // ignore any problems that may occur during error processing
                   // and continue the loop, if the server is shutting down we
                   // detect that above and cleanly exit so any other problem
                   // here is to be hardened against
                }
             }
          }
       }

       /**
        * Add the specified queue into the routing table and this router into
        * into the list of the queue's notifiables.
        *
        * @param    queue
        *           A <code>Queue</code> instance to add into routing table.
        * @param    addr
        *           The address to associate with the remote side of the queue.
        */
       private void addRoutingNode(Queue queue, int addr)
       {
          synchronized (routingTable)
          {
             routingTable.put(queue.getRemoteAddress(), queue);
             queue.setRemoteAddress(addr);
          }
       }

       /**
        * Assign an address to the node and put the node into the local node
        * registry.
        *
        * @param    queue
        *           A <code>Queue</code> instance to which address will be
        *           assigned.
        * @param    addr
        *           The address to associate with the remote side of the queue.
        */
       private void allocateAddress(Queue queue, int addr)
       {
          // is the address already set?
          if (queue.getRemoteAddress() != 0)
             return;

          // don't honor invalid addresses
          if (addr <= nodeAddress || addr > boundary)
             return;

          synchronized (leafTable)
          {
             if (!leafTable.containsKey(addr))
             {
                // this is an available address, allocate it
                leafTable.put(addr, queue);
                queue.setRemoteAddress(addr);
             }
          }
       }

       /**
        * Generate the next available address, assign it to the node and put the
        * node into the local node registry.
        *
        * @param    queue
        *           A <code>Queue</code> instance to which address will be
        *           assigned.
        */
       private void generateAddress(Queue queue)
       {
          // is the address already set?
          if (queue.getRemoteAddress() != 0)
             return;

          synchronized (leafTable)
          {
             // find and assign the next open address
             while (true)
             {
                if (lastAddress == boundary)
                {
                   // wrap back to the beginning of the address space for this
                   // node
                   lastAddress = nodeAddress + 1;
                }
                else
                {
                   // try the next possible address
                   lastAddress++;
                }

                if (!leafTable.containsKey(lastAddress))
                {
                   // this is an available address, add it
                   allocateAddress(queue, lastAddress);
                   break;
                }
             }
          }
       }

       /**
        * This method is called when a queue has been disabled, but before it has
        * been stopped.  Clean up the router's routing and leaf tables.
        *
        * @param    queue
        *           The queue that is stopping.
        */
       private void cleanUpQueue(Queue queue)
       {
          int addr = queue.getRemoteAddress();

          // the 0 address is always invalid so this means that the queue was
          // never fully initialized with an address before the queue was stopped
          // so there is nothing to clean up here
          if (addr == 0)
             return;

          if ((addr & AddressServer.NODE_MASK) != 0)
          {
             // leaf node
             synchronized (leafTable)
             {
                leafTable.remove(addr);
             }
          }
          else
          {
             // routing queue
             synchronized (routingTable)
             {
                routingTable.remove(addr);
             }
          }
       }

       /**
        * Implements a router thread which processes messages in the routing
        * queue.
        */
       private class RouterStub
       implements Runnable
       {
          /**
           * Implementation of <code>Runnable</code> interface. Just calls
           * {@link Router#route}.
           */
          public void run()
          {
             route();
          }
       }
    }