790 lines
28 KiB
Java
Executable File
790 lines
28 KiB
Java
Executable File
|
|
// 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
|