BodgeMaster
/
NORM-mirror
mirror of https://github.com/USNavalResearchLaboratory/norm.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity
NORM-mirror/examples/java/NormMsgr.java

790 lines
28 KiB
Java
Raw Blame History

// This is a Java implementation of the same
// NORM "messenger" approach as normMsgr.cpp
import java.util.concurrent.Semaphore;
import java.nio.ByteBuffer;
import java.util.Random;
import java.util.HashMap;
import java.util.LinkedList;
import mil.navy.nrl.norm.NormEvent;
import mil.navy.nrl.norm.NormInstance;
import mil.navy.nrl.norm.NormNode;
import mil.navy.nrl.norm.NormSession;
import mil.navy.nrl.norm.NormObject;
import mil.navy.nrl.norm.NormData;
import mil.navy.nrl.norm.enums.NormEventType;
import mil.navy.nrl.norm.enums.NormObjectType;
import mil.navy.nrl.norm.enums.NormAckingStatus;
import mil.navy.nrl.norm.enums.NormSyncPolicy;
public class NormMsgr
{
static final int MSG_SIZE_MAX = 65536;
static final int MSG_HDR_SIZE = 2;
private static Random randGen = new Random(System.currentTimeMillis());
// These members track NORM's "tx ready" status
private Semaphore normTxLock; // this acts as a mutex to manage "tx ready" status in a thread safe way
private Semaphore normTxReady; // this blocks data enqueuing until tx ready
private boolean norm_tx_vacancy;
private int norm_tx_queue_count;
private int norm_tx_queue_max;
private boolean norm_tx_watermark_pending;
public NormInstance normInstance;
private NormSession normSession;
private boolean norm_acking;
// This HashMap is where we cache enqueued Messages until NORM_TX_OBJECT_PURGED
// (In the C++ version we used the NormObjectSetUserData() which isn't available in NORM Java API)
private HashMap<NormData, Message> input_msg_list = new HashMap<NormData, Message>();
// Messages received from NORM are enqueued here until
// retrieved by the OutputWriter thread
private MessageQueue output_msg_queue;
private Semaphore normRxLock; // protects inter-thread access to output_msg_queue
private Semaphore normRxReady; // blocks OutputWriter when output_msg_queue is empty
// Porbably will build these congestion control modes into the NORM API
public enum NormCCMode
{
NORM_FIXED,
NORM_CC,
NORM_CCE,
NORM_CCL;
}
// Constructor
public NormMsgr() throws java.io.IOException, InterruptedException
{
normTxLock = new Semaphore(1);
normTxReady = new Semaphore(1);
try {
normInstance = new NormInstance();
normInstance.setDebugLevel(3);
}
catch (java.io.IOException ex) {
System.err.println(ex);
throw(ex);
}
// Default parameter values
norm_tx_vacancy = true;
norm_tx_queue_max = 2048; // this is set largish, since I'm testing with _small_ messages
norm_tx_queue_count = 0;
norm_tx_watermark_pending = false;
norm_acking = false;
normRxLock = new Semaphore(1);
normRxReady = new Semaphore(1);
output_msg_queue = new MessageQueue();
normRxReady.acquire(); // nothing received yet
}
public boolean openNormSession(String addr, short port, long nodeId)
{
try
{
normSession = normInstance.createSession(addr, port, nodeId);
if (null == normSession)
{
System.err.println("normMsgr error: unable to create NORM session");
return false;
}
// Set some default parameters (maybe we should put parameter setting in Start())
normSession.setRxCacheLimit(2*norm_tx_queue_max); // we let the receiver track some extra objects
normSession.setDefaultSyncPolicy(NormSyncPolicy.NORM_SYNC_ALL);
normSession.setDefaultUnicastNack(true);
normSession.setTxCacheBounds(10*1024*1024, norm_tx_queue_max, norm_tx_queue_max);
normSession.setCongestionControl(true, true);
//normSession.setMessageTrace(true);
}
catch (java.io.IOException ex)
{
System.err.println(ex);
}
return true;
} // end NormMsgr::openNormSession()
public void addAckingNode(long nodeId)
{
try {
normSession.addAckingNode(nodeId);
}
catch (java.io.IOException ex) {
System.err.println(ex);
return;
}
norm_acking = true;
} // end NormMsgr::addAckingNode()
public boolean setNormMulticastInterface(String ifaceName)
{
try {
normSession.setMulticastInterface(ifaceName);
}
catch (java.io.IOException ex) {
System.err.println(ex);
return false;
}
return true;
} // end NormMsgr::setMulticastInterface()
public void setNormCCMode(NormCCMode ccMode)
{
switch (ccMode)
{
case NORM_CC: // default TCP-friendly congestion control
normSession.setEcnSupport(false, false, false);
break;
case NORM_CCE: // "wireless-ready" ECN-only congestion control
normSession.setEcnSupport(true, true);
break;
case NORM_CCL: // "loss tolerant", non-ECN congestion control
normSession.setEcnSupport(false, false, true);
break;
case NORM_FIXED:
normSession.setEcnSupport(false, false, false);
break;
}
if (NormCCMode.NORM_FIXED != ccMode)
normSession.setCongestionControl(true);
else
normSession.setCongestionControl(false);
} // end NormMsgr::setNormCCMode()
public void setNormTxRate(double bitsPerSecond)
{
normSession.setTxRate(bitsPerSecond);
}
public void setNormDebugLevel(int level)
{normInstance.setDebugLevel(level);}
public void setNormMessageTrace(boolean state)
{normSession.setMessageTrace(state);}
public boolean start(boolean send, boolean recv)
{
// Start NORM sender and/or receiver operation
boolean recvStarted = false;
try
{
if (recv)
{
normSession.startReceiver(10*1024*1024);
recvStarted = true;
}
if (send)
{
if (norm_acking)
{
// ack-based flow control enabled on command-line,
// so disable timer-based flow control
normSession.setFlowControl(0.0);
}
// Pick a random instance id for now
int instanceId = randGen.nextInt();
normSession.startSender(instanceId, 10*1024*1024, 1400, (short)16, (short)4);
}
}
catch (java.io.IOException ex)
{
System.err.println(ex);
if (recvStarted) normSession.stopReceiver();
return false;
}
return true;
} // end NormMsgr::start()
public boolean sendMessage(Message msg)
{
// Future version will support NORM_OBJECT_STREAM as an option
while (!enqueueMessageObject(msg)); // keep trying until success (we're blocked by "normTxReady" semaphore)
return true;
} // end NormMsgr::sendMessage()
public boolean enqueueMessageObject(Message msg)
{
try
{
normTxReady.acquire(); // caller will be blocked if NORM is not "tx ready"
normTxLock.acquire(); // this guarantees protected access to "tx ready" state variables
}
catch (InterruptedException ex)
{
System.err.println(ex);
if (!normTxReady.tryAcquire())
normTxReady.release();
return false;
}
NormData obj = null;
try {
obj = normSession.dataEnqueue(msg.getBuffer(), 0, msg.getSize());
}
catch (java.io.IOException ex) {
//System.err.println(ex);
obj = null;
}
if (null == obj)
{
// Note we don't call normTxReady.release() here, it's up to the
// NormEventHandler to do that upon NORM_TX_QUEUE_EMPTY (or VACANCY for streams)
//System.err.println("NormMsgr::SendMessage() warning: data enqueue was blocked.");
norm_tx_vacancy = false; // there was no room at the inn
normTxLock.release();
return false;
}
//System.err.println("caching msg for object " + obj + "\n");
// Cache the msg associated with the resultant tx object. We use the
// input_msg_list HashMap so we can remove the msg upon NORM_TX_OBJECT_PURGED
input_msg_list.put(obj, msg);
if (norm_acking)
{
norm_tx_queue_count++;
if (!norm_tx_watermark_pending && (norm_tx_queue_count >= (norm_tx_queue_max / 2)))
{
try {
normSession.setWatermark(obj);
norm_tx_watermark_pending = true;
}
catch (java.io.IOException ex) {
System.err.println(ex);
}
}
if (norm_tx_queue_count >= norm_tx_queue_max)
{
// We've filled our tx cache, so don't call normTxReady.release()
// NormEventHandler will do this upon watermark completion
normTxLock.release();
return true;
}
}
normTxReady.release();
normTxLock.release();
return true;
}
public void onNormTxObjectPurged(NormObject normObject)
{
try {
normTxLock.acquire();
}
catch (InterruptedException ex) {
System.err.println(ex);
return; // Should kill program here (or maybe we should acquire uninterruptably)?
}
if (NormObjectType.NORM_OBJECT_DATA == normObject.getType())
{
// removed "msg" will get garbage-collected
//System.err.println("purging msg for object " + (NormData)normObject + "\n");
Message msg = input_msg_list.remove((NormData)normObject);
if (null == msg) // Shouldn't happen (and it doesn't after adding normTxLock.acquire())
System.err.println("normMsgr warning: purged invalid object?!");
}
normTxLock.release();
} // end NormMsgr::onNormTxObjectPurged()
// These next two methods are used by the NormEventHandler to update the
// NORM "tx ready" status variables in a thread-safe manner. Release of
// the "normTxReady" semaphore will unblock the InputThread if it was blocked
// due filling up the NORM tx queue.
// (The "!wasTxReady" check in these avoids any race condition with the normTxReady semaphore
public void onNormTxQueueVacancy()
{
try {
normTxLock.acquire();
}
catch (InterruptedException ex) {
System.err.println(ex);
return; // Should kill program here (maybe we should acquire uninterruptably)
}
boolean wasTxReady = norm_tx_vacancy && (norm_acking ? (norm_tx_queue_count < norm_tx_queue_max) : true);
norm_tx_vacancy = true;
boolean isTxReady = norm_acking ? (norm_tx_queue_count < norm_tx_queue_max) : true;
if (!wasTxReady && isTxReady)
{
if (normTxReady.tryAcquire())
System.err.println("NormMsgr::setNormTxVacancy() warning: normTxReady wasn't locked?!");
normTxReady.release();
}
normTxLock.release();
} // end NormMsgr::onNormTxQueueVacancy()
public void onNormTxWatermarkCompleted()
{
try {
normTxLock.acquire();
}
catch (InterruptedException ex) {
System.err.println(ex);
return; // Should kill program here (or maybe we should acquire uninterruptably)?
}
boolean wasTxReady = norm_tx_vacancy && (norm_acking ? (norm_tx_queue_count < norm_tx_queue_max) : true);
norm_tx_queue_count -= (norm_tx_queue_max / 2);
norm_tx_watermark_pending = false;
boolean isTxReady = norm_tx_vacancy && (norm_acking ? (norm_tx_queue_count < norm_tx_queue_max) : true);
if (!wasTxReady && isTxReady)
{
if (normTxReady.tryAcquire())
System.err.println("NormMsgr::decrementNormTxQueueCount() warning: normTxReady wasn't locked?!");
normTxReady.release();
}
normTxLock.release();
} // end NormMsgr::onNormTxQueueVacancy()
public void onNormRxObjectCompleted(NormObject obj)
{
try {
normRxLock.acquire();
}
catch (InterruptedException ex) {
System.err.println(ex);
return; // Should kill program here (or maybe we should acquire uninterruptably)?
}
if (NormObjectType.NORM_OBJECT_DATA == obj.getType())
{
// It's a message, so put it in the output_msg_queue for the OutputWriter thread
boolean wasEmpty = output_msg_queue.isEmpty();
Message msg = new Message(((NormData)obj).getData(), (int)obj.getSize());
output_msg_queue.append(msg);
if (wasEmpty) normRxReady.release(); // this will unblock the waiting OutputWriter
}
normRxLock.release();
} // end NormMsgr::onNormRxObjectCompleted()
// Called by OutputWriter thread to fetch received messages
public Message getRxMsg()
{
try {
normRxReady.acquire(); // will block if output_msg_queue is empty
normRxLock.acquire();
}
catch (InterruptedException ex) {
System.err.println(ex);
return null;
}
Message msg = output_msg_queue.removeHead();
if (!output_msg_queue.isEmpty())
normRxReady.release();
normRxLock.release();
if (null == msg) // shouldn't happen
System.err.println("NormMsgr warning: output_msg_queue unexpectedly empty?!");
return msg;
}
public void RunThreads()
{
InputReader inputReader = new InputReader(this);
System.err.println("main thread starting input thread ...");
inputReader.start();
NormEventHandler normEventHandler = new NormEventHandler(this);
System.err.println("main thread starting norm event handler thread ...");
normEventHandler.start();
OutputWriter outputWriter = new OutputWriter(this);
System.err.println("main thread starting output thread ...");
outputWriter.start();
// For now, we put the "inputReader" thread in the driver's seat.
// In the future, we'll be acquiring a semaphore shared among the
// threads and determine what thread(s) are signaling the parent
// and their status(es)
System.err.println("main thread waiting on input thread ...");
inputReader.acquireLock(); // when acquired, indicates thread is done
try {
inputReader.join();
normInstance.stopInstance();
normInstance.destroyInstance();
normEventHandler.join();
}
catch (InterruptedException ex) {
System.err.println(ex);
}
System.err.println("main thread exiting ...");
}
public class Message
{
private ByteBuffer msgBuffer;
// Constructors
public Message(int size)
{
msgBuffer = ByteBuffer.allocateDirect(size);
}
public Message(byte[] buffer, int size)
{
msgBuffer = ByteBuffer.wrap(buffer, 0, size);
}
public int getSize()
{return msgBuffer.capacity();}
public ByteBuffer getBuffer()
{return msgBuffer;}
} // end class NormMsgr::Message
public class MessageQueue extends LinkedList<Message>
{
public void append(Message msg)
{addLast(msg);}
public void prepend(Message msg)
{addFirst(msg);}
public Message removeHead()
{return removeFirst();}
public Message removeTail()
{return removeLast();}
public boolean isEmpty()
{
if (null == peekFirst())
return true;
else
return false;
}
} // end class NormMsgr::MessageQueue
private class InputReader extends Thread
{
private NormMsgr parentMsgr;
private boolean inputDone;
private Semaphore threadLock;
public InputReader(NormMsgr parent)
{
parentMsgr = parent;
inputDone = false;
threadLock = new Semaphore(1);
}
public boolean acquireLock()
{
try
{
threadLock.acquire();
return true;
}
catch (InterruptedException ex)
{
System.err.println(ex);
return false;
}
}
@Override
public void run()
{
System.err.println("input thread acquiring its lock ...");
try
{
threadLock.acquire();
}
catch (InterruptedException ex)
{
System.err.println(ex);
return;
}
System.err.println("input thread entering its main loop ...");
byte[] msgHeader = new byte[MSG_HDR_SIZE];
byte[] tmpBuffer = new byte[MSG_SIZE_MAX];
while (!inputDone)
{
// Read in one message at time.
int want = MSG_HDR_SIZE;
int got = 0;
while (got < want)
{
try
{
int result = System.in.read(msgHeader, got, want - got);
if (-1 == result)
{
System.err.println("normMsgr: input end-of-file!");
inputDone = true;
threadLock.release();
System.err.println("input thread exiting 1 ...");
return;
}
//System.err.println("input thread read " + result + " bytes of msg header\n");
got += result;
}
catch (java.io.IOException ex)
{
System.err.println(ex);
}
}
int msgSize = 256*msgHeader[0] + msgHeader[1];
//System.err.println("msg header is (256 * " + msgHeader[0] + ") + " + msgHeader[1] + " = " + msgSize);
got = 0;
msgSize -= 2; // already read header
while (got < msgSize)
{
try
{
int result = System.in.read(tmpBuffer, got, msgSize - got);
if (-1 == result)
{
System.err.println("normMsgr: input end-of-file!");
inputDone = true;
threadLock.release();
System.err.println("input thread exiting 2 ...");
return;
}
//System.err.println("input thread read " + result + "bytes of message data\n");
got += result;
}
catch (java.io.IOException ex)
{
System.err.println(ex);
}
}
int totalSize = msgSize + 2;
//System.err.println("normMsgr: read a " + totalSize + " byte message ...\n");
Message msg = new Message(msgSize);
msg.getBuffer().put(tmpBuffer, 0, msgSize);
parentMsgr.sendMessage(msg);
}
System.err.println("input thread releasing its lock ...");
threadLock.release();
System.err.println("input thread exiting 3 ...");
} // end InputReader::run()
} // end class NormMsgr::InputReader
private class NormEventHandler extends Thread
{
private NormMsgr parentMsgr;
public NormEventHandler(NormMsgr parent)
{
parentMsgr = parent;
}
@Override
public void run()
{
try
{
System.err.println("entering NormEventHandler loop ...");
NormEvent event;
while (null != (event = normInstance.getNextEvent()))
{
//System.err.println(event);
NormSession session = event.getSession();
switch (event.getType())
{
case NORM_TX_QUEUE_EMPTY:
case NORM_TX_QUEUE_VACANCY:
// This will unblock the InputThread if it was blocked
parentMsgr.onNormTxQueueVacancy();
break;
case NORM_TX_WATERMARK_COMPLETED:
if (NormAckingStatus.NORM_ACK_SUCCESS == session.getAckingStatus(NormNode.NORM_NODE_ANY))
{
// This will unblock the InputThread if it was blocked
//System.err.println("sender tx watermark completed ...");
parentMsgr.onNormTxWatermarkCompleted();
}
else
{
// TBD - we could see who didn't ACK and possibly remove them
// from our acking list. For now, we are infinitely
// persistent by resetting watermark ack request
session.resetWatermark();
}
break;
case NORM_TX_OBJECT_PURGED:
parentMsgr.onNormTxObjectPurged(event.getObject());
break;
case NORM_RX_OBJECT_COMPLETED:
parentMsgr.onNormRxObjectCompleted(event.getObject());
break;
case NORM_REMOTE_SENDER_INACTIVE:
// optionally free memory resources that were in
// use by this remote sender
break;
default:
break;
}
}
System.err.println("NormEventHandler got null event ...");
}
catch (java.io.IOException ex)
{
System.err.println("NormGetNextEvent IOException:");
System.err.println(ex);
}
System.err.println("exiting NormEventHandler ...");
} // end NormEventHandler::run()
} // end class NormMsgr::NormEventHandler
private class OutputWriter extends Thread
{
private NormMsgr parentMsgr;
boolean outputDone = false;
// Constructors
public OutputWriter(NormMsgr parent)
{
parentMsgr = parent;
}
@Override
public void run()
{
while (!outputDone)
{
Message msg = parentMsgr.getRxMsg(); // this will block if none available
// Make and write 2-byte length header (adding to the length to account for header, too)
int msgSize = msg.getSize() + MSG_HDR_SIZE;
byte[] msgHeader = new byte[MSG_HDR_SIZE];
// This currently assumes 2-byte MSG_HDR_SIZE
msgHeader[0] = (byte)((msgSize >> 8) & 0xff);
msgHeader[1] = (byte)(msgSize & 0xff);
System.out.write(msgHeader, 0, MSG_HDR_SIZE);
// Write message content
System.out.write(msg.getBuffer().array(), 0, msg.getSize());
}
System.err.println("NormMsgr output thread exiting ...");
}
} // end class NormMsgr::OutputWriter()
public static void usage()
{
System.err.println("Usage: normMsgr id <nodeId> {send &| recv} [addr <addr>[/<port>]][ack <node1>[,<node2>,...]\n" +
" [cc|cce|ccl|rate <bitsPerSecond>][interface <name>][debug <level>][trace]\n");
}
public static void main(String[] args) throws Throwable
{
// Default parameters
String sessionAddr = "224.1.2.3";
short sessionPort = 6003;
long nodeId = 0;
boolean send = false;
boolean recv = false;
String[] ackerList = null;
String mcastIface = null;
int debugLevel = 0;
boolean normTrace = false;
NormCCMode ccMode = NormCCMode.NORM_CC;
double txRate = 1.0e+06; // only applies for NORM_FIXED ccMode
int i = 0;
while (i < args.length)
{
String cmd = args[i++];
if (cmd.equals("id"))
{
nodeId = Integer.parseInt(args[i++]);
}
else if (cmd.equals("send"))
{
send = true;
}
else if (cmd.equals("recv"))
{
recv = true;
}
else if (cmd.equals("addr"))
{
String[] items = args[i++].split("/");
sessionAddr = items[0];
if (items.length > 1)
sessionPort = (short)Integer.parseInt(items[1]);
}
else if (cmd.equals("interface"))
{
mcastIface = args[i++];
}
else if (cmd.equals("cce"))
{
ccMode = NormCCMode.NORM_CCE;
}
else if (cmd.equals("cce"))
{
ccMode = NormCCMode.NORM_CCL;
}
else if (cmd.equals("rate"))
{
ccMode = NormCCMode.NORM_FIXED;
txRate = Double.parseDouble(args[i++]);
}
else if (cmd.equals("ack"))
{
ackerList = args[i++].split(",");
}
else if (cmd.equals("debug"))
{
debugLevel = Integer.parseInt(args[i++]);
}
else if (cmd.equals("trace"))
{
normTrace = true;
}
else
{
System.err.println("normMsgr error: invalid command '" + cmd + "'");
usage();
return;
}
}
if (!send && !recv)
{
System.err.println("normMsgr error: not configured to send or recv!");
usage();
return;
}
if (0 == nodeId)
{
System.err.println("normMsgr error: no local 'id' provided!");
usage();
return;
}
NormMsgr msgr = new NormMsgr();
msgr.setNormDebugLevel(debugLevel);
msgr.openNormSession(sessionAddr, sessionPort, nodeId);
if (null != mcastIface)
msgr.setNormMulticastInterface(mcastIface);
if (null != ackerList)
{
for (i = 0; i < ackerList.length; i++)
msgr.addAckingNode(Integer.parseInt(ackerList[i]));
}
msgr.setNormCCMode(ccMode);
if (NormCCMode.NORM_FIXED == ccMode)
msgr.setNormTxRate(txRate);
msgr.setNormMessageTrace(normTrace);
msgr.start(send, recv);
msgr.RunThreads();
System.err.println("normMsgr: main() exiting ...");
}
} // end class NormMsgr
Reference in New Issue View Git Blame Copy Permalink