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();
}
}
}