Protocol.java
/*
** Module : Protocol.java
** Abstract : low level driver of the P2J protocol
**
** Copyright (c) 2005-2024, Golden Code Development Corporation.
**
** -#- -I- --Date-- --JPRM-- ----------------Description-----------------
** 001 SIY 20050120 @19501 Created initial version.
** 002 SIY 20050207 @19712 Implemented exception delivering scheme.
** Also modified stop() to avoid waiting for
** threads.
** 003 SIY 20050211 @19784 Reset stream to exclude backreferencing.
** 004 NVS 20050308 @20250 Using NetSocket class instead of SSLSocket.
** The ObjectInput and ObjectOutput streams are
** created in NetSocket.
** 005 SIY 20050315 @20385 Organized imports and other minor cleanups.
** Added call to stream flush() method (merge
** from NVS).
** 006 SIY 20050329 @20546 Minor changes in member access rights to
** improve performance (change is very small).
** 007 GES 20060211 @24533 Minor cleanup and added to-dos to exception
** processing.
** 008 NVS 20060616 @27376 Exception processing modified. No more stack
** trace printing.
** 009 GES 20060726 @28199 Added support for bidirectional state
** synchronization.
** 010 GES 20060731 @28256 Greatly improved performance by reducing the
** overhead of object graph serialization. The
** trick is to serialize the object twice! First
** the object is flattened via serialization to
** a byte[]. Then this byte[] is serialized to
** the socket. This intermediate step is
** significantly faster than serializing direct
** to the socket! The reading was modified to
** handle the proper restore.
** 011 GES 20060802 @28293 Remove attachments before enqueueing because
** REQUEST_SYNC messages tend to reuse the same
** message on the return trip, but just swap
** addresses. Thus the same state sync data was
** sent back when it didn't need to be in the
** case where the attachment would have been
** null (no changes to send).
** 012 NVS 20061017 @30464 Implemented ability to restart the session
** without exiting the process.
** 013 GES 20061207 @31612 Added logging and isRunning().
** 014 NVS 20070411 @32929 Protocol threads are created as part of a
** thread group named "protocol".
** 015 ECF 20071118 @35884 Reworked exception handling in reader and
** writer threads.
** 016 GES 20081024 @40328 Eliminated use of the NetSocket/Socket and
** instead now the protocol is only dependent
** upon Queue APIs. This reduces dependencies
** between components and fosters better
** encapsulation of the protocol.
** 017 ECF 20090205 @42106 Changed to accommodate updated Queue API.
** Queue.stop() method now accepts a boolean
** parameter to optionally switch security
** context to the session's user's context.
** 018 GES 20090722 @43334 Better thread names to enhance debugging.
** 019 GES 20090723 @43346 Better handling of OOME on reader/writer threads.
** Improved control over logging detail. Synchronized
** reader/writer thread termination processing to
** avoid extra errors occurring do to a race
** condition.
** 020 CA 20090902 @43821 Changed logging level for any SocketException's
** caught while the queue is still running to always
** use FINE.
** 021 CA 20131121 The sync'ed state processing is moved from Reader/Writer threads to
** the thread which actually sends/receives the message (i.e.
** Conversation thread).
** 022 SBI 20151119 Added stopQueue with SilentUnwindException checking.
** 023 CA 20190315 No state synchronization is performed for async replies or requests.
** 024 IAS 20200722 Refactored to the new NetSocket API.
** 025 CA 20200923 Improved performance for objToByteArray and byteArrayToObj - use Externalizable.
** Changed StateSynchronizer to return Externalizable instances.
** 026 IAS 20200930 Added JMX counters.
** 027 IAS 20201001 Ged rid of byte arrays allocations in objToByteArray().
** IAS 20210325 Added SSLException catch
** TJD 20220607 Writer/ByteArrayOutputStream memory leak fix for LOBs
** 028 GBB 20230512 Logging methods replaced by CentralLogger/ConversionStatus.
** 029 GBB 20230620 createThreadName method moved to Utils.
** 030 HC 20240222 Enabled JMX on FWD Client.
** 031 CA 20240809 Skip using JMX timers when JMX_DEBUG flag is not set.
*/
/*
** 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.io.*;
import java.net.*;
import java.util.logging.*;
import javax.net.ssl.*;
import com.goldencode.p2j.jmx.*;
import com.goldencode.p2j.util.*;
import com.goldencode.p2j.util.logging.*;
/**
* This class implements the low level part of the protocol. It provides two
* threads (Reader and Writer). Reader thread instantiates messages from the
* stream and passes to the queue for the further processing. Writer thread
* stores messages to stream. Also, <code>Protocol</code> is responsible for
* notifying queue about I/O errors.
* <p>
* The reading and writing loops implement the proper callbacks for the
* {@link StateSynchronizer} implementation of bidirectional state
* synchronization. This processing can be activated using the
* <code>registerSynchronizer</code> method.
* <p>
* WARNING: the core loops of the reader and writer threads are controlled
* by a variable which they read outside of a synchronized block. This means
* that it is possible (depending on the JVM implementation) to get
* inconsistent or corrupt results because it may be read in the middle of
* being changed. At this time, we are not introducing a lock object since
* this would cause potential contention to occur between the reader and
* writer threads. This is something that needs to be tested to ensure its
* impact is understood but it DOES need to be fixed.
*/
class Protocol
{
/** Counter for measuring incoming traffic */
private static final SimpleCounter INCOMING = SimpleCounter.getInstance(
FwdServerJMX.Counter.NetworkReads
);
/** Counter for measuring outgoing traffic */
private static final SimpleCounter OUTGOING = SimpleCounter.getInstance(
FwdServerJMX.Counter.NetworkWrites
);
/** timer for measuring objToByteArray time */
private static final NanoTimer TO_BYTE_ARRAY_TIMER = NanoTimer.getInstance(
FwdServerJMX.TimeStat.objToByteArray
);
/** Logger. */
private static final CentralLogger LOG = CentralLogger.get(Protocol.class.getName());
/** Constant identifying a {@link Message} instance. */
private static final int TYPE_MESSAGE = 0;
/** Constant identifying a {@link SyncMessage} instance. */
private static final int TYPE_SYNC_MESSAGE = 1;
/** Thread group for the protocol threads. */
private static ThreadGroup protoThreads = new ThreadGroup(
Thread.currentThread().getThreadGroup(), "protocol");
/** Flag indicating status for Reader/Writer threads. */
private volatile boolean running = false;
/** The queue representing the session with which protocol is working. */
private Queue queue = null;
/** Reader thread. */
private Thread readerThread = null;
/** Writer thread. */
private Thread writerThread = null;
/** Synchronizes access to the state synchronizer. */
private Object slock = new Object();
/** The object used to implement state synchronization if enabled. */
private StateSynchronizer ssync = null;
/** Synchronizes termination processing. */
private Object tlock = new Object();
/**
* Constructs an instance using the provided queue.
*
* @param queue
* Instance of the <code>Queue</code> to/from which messages
* are written/read.
*/
Protocol(Queue queue)
{
this.queue = queue;
}
/**
* Starts and Writer threads as daemon threads. This occurs so
* long as:
* <ul>
* <li> The threads are not already started.
* <li> The queue was properly initialized.
* <li> The low-level network transport is not closed.
* </ul>
*/
synchronized void start()
{
if (running == true || queue == null || queue.isTransportClosed())
return;
running = true;
Runnable reader = new Reader();
Runnable writer = new Writer();
readerThread = new Thread(protoThreads, reader, "Reader");
readerThread.setDaemon(true);
writerThread = new Thread(protoThreads, writer, "Writer");
writerThread.setDaemon(true);
readerThread.start();
writerThread.start();
}
/**
* Stops Reader and Writer threads.
*/
synchronized void stop()
{
if (!running)
return;
running = false;
readerThread.interrupt();
writerThread.interrupt();
queue.closeTransport();
readerThread = null;
writerThread = null;
}
/**
* Force the reader and writer threads to have more descriptive names
* to aid debugging. This is only meaningful on a router node.
*/
synchronized void setNames()
{
SessionManager sessMgr = SessionManager.get();
if (sessMgr.isRouter())
{
String rname = Utils.createThreadName("Reader");
String wname = Utils.createThreadName("Writer");
if (rname != null)
readerThread.setName(rname);
if (wname != null)
writerThread.setName(wname);
}
}
/**
* Returns the operational state of this protocol driver.
*
* @return <code>true</code> if the protocol is running (operational)
* and <code>false</code> if the session is closed.
*/
public synchronized boolean isRunning()
{
return running;
}
/**
* Sets the instance of the state synchronizer the reader and writer
* threads should use, if and only if there is no other state synchronizer
* already in use. This "there can BE only ONE" approach can be changed
* easily in the future but was implemented to ensure high performance
* since each loop of the protocol must call the synchronization
* primitives.
*
* @param ssync
* The synchronizer to register or <code>null</code> if the
* currently registered synchronizer should be deregistered.
*/
void registerSynchronizer(StateSynchronizer ssync)
{
synchronized (slock)
{
if (this.ssync == null || ssync == null)
{
this.ssync = ssync;
}
}
}
/**
* Checks if there is a registered state synchronizer and if so, it is
* queried to obtain the latest changes in state that must be sent to the
* other side of the connection. If any changes are found they are
* packaged into the message and returned, otherwise the original message
* is returned. If no state synchronizer is registered, the original
* message is returned.
* <p>
* The state synchronization is allowed only for conversation mode or other processing which
* is known to be synchronous.
* <p>
* For async replies or requests, no state services are used, as the state is something which
* is required to be the same (and in sync) between both sides. The very nature of async
* messages may make the state to be inconsistent between the running threads.
*
* @param outbound
* The message that was dequeued for transmission.
*
* @return The updated message (with attached state) if there is a
* registered state synchronizer AND there are state changes
* to send. Otherwise the original message will be returned.
*/
Message attachChanges(Message outbound)
{
if (outbound.isAsyncReply() || outbound.isAsyncRequest())
{
// do not attach changes if the message is async - as this can get deadlocked with other
// Conversation-thread processing.
return outbound;
}
Externalizable state = null;
Message result = outbound;
synchronized (slock)
{
if (ssync != null)
{
state = ssync.getChanges();
}
}
if (state != null)
{
result = new SyncMessage(outbound, state);
}
return result;
}
/**
* If there is a registered state synchronizer and if there are any
* attached state updates sent from the other side of the connection,
* these updates are passed to the state synchronizer to be applied to
* this node's state.
*
* @param inbound
* The message that is about to be enqueued for processing.
*
* @return An instance of the message without an attachment or the
* given message (unchanged) if there was no attachment.
*/
Message applyChanges(Message inbound)
{
if (inbound instanceof SyncMessage)
{
SyncMessage smsg = (SyncMessage) inbound;
Externalizable state = smsg.getState();
synchronized (slock)
{
if (ssync != null)
{
ssync.applyChanges(state);
}
}
// remove the attachment (this copy constructor knows nothing of
// the subclass so we will only get a simple message as a result)
inbound = new Message(inbound);
}
return inbound;
}
/**
* Stop the queue with SilentUnwindException checking.
*
* @param result
* The managed exception.
*/
private void stopQueue(Exception result)
{
try
{
if (running)
{
queue.stop(result, true);
}
}
catch (SilentUnwindException ex)
{
LOG.log(Level.WARNING, "stopQueue", ex);
}
}
/**
* This helper class implements functionality of Reader thread.
*/
private class Reader
implements Runnable
{
/**
* This method instantiates messages from the stream and places into
* the queue for further processing.
*/
public void run()
{
Exception result = null;
try
{
while (running)
{
// read the next message from the stream
byte[] data = queue.read();
if (data == null)
{
continue;
}
INCOMING.update(data.length);
// re-inflate the flattened object that came in as a byte[]
// (this is faster than without the intermediate flattened
// form)
Message msg = byteArrayToObj(data);
// we can already release data memory here
data = null;
// hand the message to the dispatcher for processing
queue.enqueueInbound(msg);
}
}
catch (EOFException exc)
{
// this happens when the peer side terminates so this is NOT
// a form of unexpected failure
}
catch (InterruptedException exc)
{
// an interrupted exception will be caused by the stop() method
// so we consider this a "clean" way of exiting
}
catch (SocketException | SSLException exc)
{
// this happens when the socket is closed so this is NOT a form
// of unexpected failure if the running flag was set to false
if (running)
{
// log that there was an unexpected failure
if (LOG.isLoggable(Level.FINE))
{
LOG.logp(Level.FINE,
"Protocol.Reader.run()",
"",
"Peer disconnected!",
exc);
}
result = exc;
}
}
catch (Throwable thr)
{
// unexpected failure
LOG.logp(Level.WARNING,
"Protocol.Reader.run()",
"",
"failure in reading loop",
thr);
result = (thr instanceof Exception) ? (Exception) thr
: new RuntimeException(thr);
}
finally
{
// only one of the (reader and writer) threads should process
// termination at any given time; the first one in gets to process
// and the other will be blocked until the processing is complete;
synchronized (tlock)
{
// the second thread will find that the running flag is cleared
// so it won't need to and won't try to execute termination
// processing
stopQueue(result);
}
}
}
/**
* Restore ("inflate") an object from an in-memory array of bytes.
*
* @param obj
* The flattened object to restore.
*
* @return The inflated object.
* @throws IOException
* on error
* @throws ClassNotFoundException
* if class specified in the binary stream was not found
*/
private Message byteArrayToObj(byte[] obj)
throws IOException,
ClassNotFoundException
{
Message msg = obj[0] == TYPE_MESSAGE ? new Message() : new SyncMessage();
ByteArrayInputStream bais = new ByteArrayInputStream(obj, 1, obj.length);
ObjectInputStream ois = new ObjectInputStream(bais);
msg.readExternal(ois);
return msg;
}
}
/**
* Custom version of the ByteArrayOutputStream with access to underlying buffer
*
*/
public class ByteArrayOutputStream2 extends ByteArrayOutputStream
{
/**
* Constructor
*/
public ByteArrayOutputStream2()
{
super();
}
/**
* Constructor
* @param size
* initial capacity
*/
public ByteArrayOutputStream2(int size)
{
super(size);
}
/**
* Returns the internal buffer of this ByteArrayOutputStream, without copying.
* @return the internal buffer
*/
public synchronized byte[] buf() {
return this.buf;
}
}
/**
* This helper class implements functionality of Writer thread.
*/
private class Writer
implements Runnable
{
/**
* Default size of buffer for object flattening
*/
private static final int BUFFER_SIZE = 8192;
/**
* Pre-allocated helper buffers for object flattening
*/
private ByteArrayOutputStream2 baos = new ByteArrayOutputStream2(BUFFER_SIZE);
private ByteArrayOutputStream res = new ByteArrayOutputStream(BUFFER_SIZE);
/**
* Holder for data to be written out
*/
private byte[] data;
/**
* Set new value of the <code>data</code> field (by be used in lambdas)
* @param data
* new value of the field
*/
private void setData(byte[] data)
{
this.data = data;
}
/**
* This method obtains messages from the Outbound Queue and writes them
* into output stream.
*/
public void run()
{
Exception result = null;
try
{
while (running)
{
// dequeue a message
Message obj = queue.dequeueOutbound();
// flatten the object to a byte[] before writing it to the
// socket since this is significantly faster than going
// without the intermediate byte[]; note that we still use
// serialization for the final output since the low level
// Java approach handles reading an entire array as a single
// entity which is something we want to avoid doing ourselves
if (FwdServerJMX.JMX_DEBUG)
{
TO_BYTE_ARRAY_TIMER.timer(() -> setData(objToByteArray(obj)));
}
else
{
setData(objToByteArray(obj));
}
if (data != null)
{
OUTGOING.update(data.length);
}
queue.write(data);
// allow GC to collect memory allocated for data
data = null;
}
}
catch (InterruptedException exc)
{
// an interrupted exception will be caused by the stop() method
// so we consider this a "clean" way of exiting
}
catch (SocketException exc)
{
// this happens when the socket is closed so this is NOT a form
// of unexpected failure if the running flag was set to false
if (running)
{
// log that there was an unexpected failure
if (LOG.isLoggable(Level.FINE))
{
LOG.logp(Level.FINE,
"Protocol.Writer.run()",
"",
"Peer disconnected!",
exc);
}
result = exc;
}
}
catch (Throwable thr)
{
// unexpected failure
LOG.logp(Level.WARNING,
"Protocol.Writer.run()",
"",
"failure in writing loop",
thr);
result = (thr instanceof Exception) ? (Exception) thr
: new RuntimeException(thr);
}
finally
{
// only one of the (reader and writer) threads should process
// termination at any given time; the first one in gets to process
// and the other will be blocked until the processing is complete;
synchronized (tlock)
{
// the second thread will find that the running flag is cleared
// so it won't need to and won't try to execute termination
// processing
stopQueue(result);
}
}
}
/**
* Serialize the given object into an in-memory array of bytes.
*
* @param obj
* The graph of objects to be converted.
*
* @return The "flattened" version.
* @throws IOException
* on error
*/
private byte[] objToByteArray(Message obj)
throws IOException
{
baos.reset();
ObjectOutputStream oos = new ObjectOutputStream(baos);
obj.writeExternal(oos);
oos.flush();
res.reset();
res.write(obj.getClass() == Message.class ? TYPE_MESSAGE : TYPE_SYNC_MESSAGE);
res.write(baos.buf(), 0, baos.size());
// in case baos grows too big -> reallocate it
if (baos.size() > BUFFER_SIZE)
{
baos = new ByteArrayOutputStream2(BUFFER_SIZE);
}
byte[] result = res.toByteArray();
// in case res grows too big -> reallocate it
if (res.size() > BUFFER_SIZE)
{
res = new ByteArrayOutputStream(BUFFER_SIZE);
}
return result;
}
}
}