diff --git a/RELEASE.md b/RELEASE.md index 10f91775..9b30ae15 100644 --- a/RELEASE.md +++ b/RELEASE.md @@ -1,4 +1,4 @@ #### Port Aeron.NET has been ported against Java version: -- Agrona: 0.9.18-13-g0378ffa -- Aeron: 1.10.2 +- Agrona: 0.9.22 +- Aeron: 1.10.4 diff --git a/driver/Aeron.Driver.nuspec b/driver/Aeron.Driver.nuspec index 0974dd12..a73df3a6 100644 --- a/driver/Aeron.Driver.nuspec +++ b/driver/Aeron.Driver.nuspec @@ -2,7 +2,7 @@ Aeron.Driver - 1.10.3 + 1.10.4 Aeron Driver Adaptive Financial Consulting Ltd. Adaptive Financial Consulting Ltd. diff --git a/driver/media-driver.jar b/driver/media-driver.jar index 61852b79..b6d94a26 100644 Binary files a/driver/media-driver.jar and b/driver/media-driver.jar differ diff --git a/driver/version.txt b/driver/version.txt index 6e7b9df6..37bf5ee2 100644 --- a/driver/version.txt +++ b/driver/version.txt @@ -1,2 +1,2 @@ Driver source: -http://repo1.maven.org/maven2/io/aeron/aeron-all/1.10.3/aeron-all-1.10.3.jar +http://repo1.maven.org/maven2/io/aeron/aeron-all/1.10.4/aeron-all-1.10.4.jar diff --git a/src/Adaptive.Aeron/Adaptive.Aeron.csproj b/src/Adaptive.Aeron/Adaptive.Aeron.csproj index 5139b7a7..be4c7e12 100644 --- a/src/Adaptive.Aeron/Adaptive.Aeron.csproj +++ b/src/Adaptive.Aeron/Adaptive.Aeron.csproj @@ -2,7 +2,7 @@ netstandard2.0;net45 Aeron.Client - 1.10.3 + 1.10.4 Adaptive Financial Consulting Ltd. Adaptive Financial Consulting Ltd. Aeron Client diff --git a/src/Adaptive.Aeron/Aeron.cs b/src/Adaptive.Aeron/Aeron.cs index cc19bf0f..1150fb9a 100644 --- a/src/Adaptive.Aeron/Aeron.cs +++ b/src/Adaptive.Aeron/Aeron.cs @@ -526,6 +526,11 @@ public class Context : IDisposable /// Key for the tags for a channel /// public const string TAGS_PARAM_NAME = "tags"; + + /// + /// Parameter name for channel URI param to indicate if term buffers should be sparse. Value is boolean. + /// + public const string SPARSE_PARAM_NAME = "sparse"; /// /// Get the default directory name to be used if is not set. This will take @@ -1418,12 +1423,16 @@ public int SaveErrorLog(StreamWriter writer, MappedByteBuffer cncByteBuffer) cncVersion); } + int distinctErrorCount = 0; UnsafeBuffer buffer = CncFileDescriptor.CreateErrorLogBuffer(cncByteBuffer, cncMetaDataBuffer); - void ErrorConsumer(int count, long firstTimestamp, long lastTimestamp, string ex) - => FormatError(writer, count, firstTimestamp, lastTimestamp, ex); + if (ErrorLogReader.HasErrors(buffer)) + { + void ErrorConsumer(int count, long firstTimestamp, long lastTimestamp, string ex) + => FormatError(writer, count, firstTimestamp, lastTimestamp, ex); - var distinctErrorCount = ErrorLogReader.Read(buffer, ErrorConsumer); + distinctErrorCount = ErrorLogReader.Read(buffer, ErrorConsumer); + } writer.WriteLine(); writer.WriteLine("{0} distinct errors observed.", distinctErrorCount); diff --git a/src/Adaptive.Aeron/ChannelUriStringBuilder.cs b/src/Adaptive.Aeron/ChannelUriStringBuilder.cs index bda05484..65566287 100644 --- a/src/Adaptive.Aeron/ChannelUriStringBuilder.cs +++ b/src/Adaptive.Aeron/ChannelUriStringBuilder.cs @@ -24,6 +24,7 @@ public class ChannelUriStringBuilder private string _controlMode; private string _tags; private bool? _reliable; + private bool? _sparse; private int? _ttl; private int? _mtu; private int? _termLength; @@ -272,6 +273,30 @@ public ChannelUriStringBuilder Reliable(bool? isReliable) return _reliable; } + /// + /// Set to indicate if a term log buffer should be sparse on disk or not. Sparse saves space at the potential + /// expense of latency. + /// + /// true if the term buffer log is sparse on disk. + /// this for a fluent API. + /// + public ChannelUriStringBuilder Sparse(bool? isSparse) + { + _sparse = isSparse; + return this; + } + + /// + /// Get if a term log buffer should be sparse on disk or not. Sparse saves space at the potential expense of latency. + /// + /// true if the term buffer log is sparse on disk. + /// + public bool? Sparse() + { + return _sparse; + } + + /// /// Set the Time To Live (TTL) for a multicast datagram. Valid values are 0-255 for the number of hops the datagram /// can progress along. @@ -603,6 +628,11 @@ public string Build() _sb.Append(Aeron.Context.RELIABLE_STREAM_PARAM_NAME).Append('=').Append(_reliable).Append('|'); } + if (null != _sparse) + { + _sb.Append(Aeron.Context.SPARSE_PARAM_NAME).Append('=').Append(_sparse).Append('|'); + } + if (null != _ttl) { _sb.Append(Aeron.Context.TTL_PARAM_NAME).Append('=').Append(_ttl.Value).Append('|'); diff --git a/src/Adaptive.Aeron/Publication.cs b/src/Adaptive.Aeron/Publication.cs index 0be5c1c0..2d143264 100644 --- a/src/Adaptive.Aeron/Publication.cs +++ b/src/Adaptive.Aeron/Publication.cs @@ -71,7 +71,7 @@ public abstract class Publication : IDisposable /// The offer failed due to reaching the maximum position of the stream given term buffer length times the total /// possible number of terms. /// - /// If this happen then the publication should be closed and a new one added. To make it less likely to happen then + /// If this happens then the publication should be closed and a new one added. To make it less likely to happen then /// increase the term buffer length. /// /// diff --git a/src/Adaptive.Agrona/Adaptive.Agrona.csproj b/src/Adaptive.Agrona/Adaptive.Agrona.csproj index ce26d7b0..517001d3 100644 --- a/src/Adaptive.Agrona/Adaptive.Agrona.csproj +++ b/src/Adaptive.Agrona/Adaptive.Agrona.csproj @@ -3,7 +3,7 @@ netstandard2.0;net45 true Agrona - 1.10.3 + 1.10.4 Adaptive Financial Consulting Ltd. Adaptive Financial Consulting Ltd. Agrona libraries initially included in Aeron Client diff --git a/src/Adaptive.Agrona/Concurrent/Errors/ErrorReader.cs b/src/Adaptive.Agrona/Concurrent/Errors/ErrorReader.cs index 45396647..d758add4 100644 --- a/src/Adaptive.Agrona/Concurrent/Errors/ErrorReader.cs +++ b/src/Adaptive.Agrona/Concurrent/Errors/ErrorReader.cs @@ -23,6 +23,16 @@ namespace Adaptive.Agrona.Concurrent.Errors /// public class ErrorLogReader { + /// + /// Has the error buffer any recorded errors? + /// + /// containing the . + /// true if there is at least one error. + public static bool HasErrors(IAtomicBuffer buffer) + { + return 0 != buffer.GetIntVolatile(DistinctErrorLog.LengthOffset); + } + /// /// Read all the errors in a log since the creation of the log. /// diff --git a/src/Adaptive.Archiver/Adaptive.Archiver.csproj b/src/Adaptive.Archiver/Adaptive.Archiver.csproj index ce9bf293..53b4a30d 100644 --- a/src/Adaptive.Archiver/Adaptive.Archiver.csproj +++ b/src/Adaptive.Archiver/Adaptive.Archiver.csproj @@ -3,7 +3,7 @@ netstandard2.0;net45 true Aeron.Archiver - 1.10.3 + 1.10.4 Adaptive Financial Consulting Ltd. Adaptive Financial Consulting Ltd. Archiving over the Aeron transport diff --git a/src/Adaptive.Archiver/AeronArchive.cs b/src/Adaptive.Archiver/AeronArchive.cs index aa0c3b87..56651c46 100644 --- a/src/Adaptive.Archiver/AeronArchive.cs +++ b/src/Adaptive.Archiver/AeronArchive.cs @@ -301,7 +301,8 @@ public virtual void CheckForErrorResponse() if (controlResponsePoller.TemplateId() == ControlResponseDecoder.TEMPLATE_ID && controlResponsePoller.Code() == ControlResponseCode.ERROR) { - throw new ArchiveException(controlResponsePoller.ErrorMessage(), (int) controlResponsePoller.RelevantId()); + throw new ArchiveException(controlResponsePoller.ErrorMessage(), + (int) controlResponsePoller.RelevantId()); } } } @@ -1066,6 +1067,16 @@ public class Configuration /// public const int RECORDING_EVENTS_STREAM_ID_DEFAULT = 30; + /// + /// Sparse term buffer indicator for control streams. + /// + private const string CONTROL_TERM_BUFFER_SPARSE_PARAM_NAME = "aeron.archive.control.term.buffer.sparse"; + + /// + /// Overrides driver's sparse term buffer indicator for control streams. + /// + private const bool CONTROL_TERM_BUFFER_SPARSE_DEFAULT = true; + /// /// Term length for control streams. /// @@ -1077,12 +1088,12 @@ public class Configuration internal const int CONTROL_TERM_BUFFER_LENGTH_DEFAULT = 64 * 1024; /// - /// Term length for control streams. + /// MTU length for control streams. /// internal const string CONTROL_MTU_LENGTH_PARAM_NAME = "aeron.archive.control.mtu.length"; /// - /// MTU to reflect default control term length. + /// MTU to reflect default for the control streams. /// internal const int CONTROL_MTU_LENGTH_DEFAULT = 4 * 1024; @@ -1096,11 +1107,22 @@ public static long MessageTimeoutNs() return Config.GetDurationInNanos(MESSAGE_TIMEOUT_PROP_NAME, MESSAGE_TIMEOUT_DEFAULT_NS); } + /// + /// Should term buffer files be sparse for control request and response streams. + /// + /// true if term buffer files should be sparse for control request and response streams. + /// + public static bool ControlTermBufferSparse() + { + string propValue = Config.GetProperty(CONTROL_TERM_BUFFER_SPARSE_PARAM_NAME); + return null != propValue ? "true".Equals(propValue) : CONTROL_TERM_BUFFER_SPARSE_DEFAULT; + } + /// /// Term buffer length to be used for control request and response streams. /// /// term buffer length to be used for control request and response streams. - /// + /// public static int ControlTermBufferLength() { return Config.GetSizeAsInt(CONTROL_TERM_BUFFER_LENGTH_PARAM_NAME, CONTROL_TERM_BUFFER_LENGTH_DEFAULT); @@ -1217,6 +1239,7 @@ public class Context : IDisposable internal int controlRequestStreamId = Configuration.ControlStreamId(); internal string controlResponseChannel = Configuration.ControlResponseChannel(); internal int controlResponseStreamId = Configuration.ControlResponseStreamId(); + internal bool controlTermBufferSparse = Configuration.ControlTermBufferSparse(); internal int controlTermBufferLength = Configuration.ControlTermBufferLength(); internal int controlMtuLength = Configuration.ControlMtuLength(); @@ -1260,6 +1283,7 @@ public void Conclude() ChannelUri uri = ChannelUri.Parse(controlRequestChannel); uri.Put(Aeron.Aeron.Context.TERM_LENGTH_PARAM_NAME, Convert.ToString(controlTermBufferLength)); uri.Put(Aeron.Aeron.Context.MTU_LENGTH_PARAM_NAME, Convert.ToString(controlMtuLength)); + uri.Put(Aeron.Aeron.Context.SPARSE_PARAM_NAME, Convert.ToString(controlTermBufferSparse)); controlRequestChannel = uri.ToString(); } @@ -1418,9 +1442,31 @@ public int ControlResponseStreamId() { return controlResponseStreamId; } + + /// + /// Should the control streams use sparse file term buffers. + /// + /// for the control stream. + /// this for a fluent API. + /// + public Context ControlTermBufferSparse(bool controlTermBufferSparse) + { + this.controlTermBufferSparse = controlTermBufferSparse; + return this; + } + + /// + /// Should the control streams use sparse file term buffers. + /// + /// true if the control stream should use sparse file term buffers. + /// + public bool ControlTermBufferSparse() + { + return controlTermBufferSparse; + } /// - /// Set the term buffer length for the control stream. + /// Set the term buffer length for the control streams. /// /// for the control stream. /// this for a fluent API. @@ -1432,9 +1478,9 @@ public Context ControlTermBufferLength(int controlTermBufferLength) } /// - /// Get the term buffer length for the control steam. + /// Get the term buffer length for the control streams. /// - /// the term buffer length for the control steam. + /// the term buffer length for the control streams. /// public int ControlTermBufferLength() { @@ -1442,9 +1488,9 @@ public int ControlTermBufferLength() } /// - /// Set the MTU length for the control stream. + /// Set the MTU length for the control streams. /// - /// for the control stream. + /// for the control streams. /// this for a fluent API. /// public Context ControlMtuLength(int controlMtuLength) @@ -1454,9 +1500,9 @@ public Context ControlMtuLength(int controlMtuLength) } /// - /// Get the MTU length for the control steam. + /// Get the MTU length for the control steams. /// - /// the MTU length for the control steam. + /// the MTU length for the control steams. /// public int ControlMtuLength() { @@ -1648,7 +1694,8 @@ public AeronArchive Poll() if (2 == step) { - if (!archiveProxy.TryConnect(ctx.ControlResponseChannel(), ctx.ControlResponseStreamId(), connectCorrelationId)) + if (!archiveProxy.TryConnect(ctx.ControlResponseChannel(), ctx.ControlResponseStreamId(), + connectCorrelationId)) { return null; } @@ -1667,14 +1714,17 @@ public AeronArchive Poll() } controlResponsePoller.Poll(); - if (controlResponsePoller.IsPollComplete() && controlResponsePoller.CorrelationId() == connectCorrelationId && controlResponsePoller.TemplateId() == ControlResponseDecoder.TEMPLATE_ID) + if (controlResponsePoller.IsPollComplete() && + controlResponsePoller.CorrelationId() == connectCorrelationId && + controlResponsePoller.TemplateId() == ControlResponseDecoder.TEMPLATE_ID) { ControlResponseCode code = controlResponsePoller.Code(); if (code != ControlResponseCode.OK) { if (code == ControlResponseCode.ERROR) { - throw new ArchiveException("error: " + controlResponsePoller.ErrorMessage(), (int) controlResponsePoller.RelevantId()); + throw new ArchiveException("error: " + controlResponsePoller.ErrorMessage(), + (int) controlResponsePoller.RelevantId()); } throw new ArchiveException("unexpected response: code=" + code); @@ -1682,7 +1732,9 @@ public AeronArchive Poll() long controlSessionId = controlResponsePoller.ControlSessionId(); Subscription subscription = controlResponsePoller.Subscription(); - return new AeronArchive(ctx, controlResponsePoller, archiveProxy, new RecordingDescriptorPoller(subscription, FRAGMENT_LIMIT, controlSessionId), controlSessionId); + return new AeronArchive(ctx, controlResponsePoller, archiveProxy, + new RecordingDescriptorPoller(subscription, FRAGMENT_LIMIT, controlSessionId), + controlSessionId); } return null; diff --git a/src/Adaptive.Cluster/Adaptive.Cluster.csproj b/src/Adaptive.Cluster/Adaptive.Cluster.csproj index de1967ab..e6168d05 100644 --- a/src/Adaptive.Cluster/Adaptive.Cluster.csproj +++ b/src/Adaptive.Cluster/Adaptive.Cluster.csproj @@ -3,7 +3,7 @@ netstandard2.0;net45 true Aeron.Cluster - 1.10.3 + 1.10.4 Adaptive Financial Consulting Ltd. Adaptive Financial Consulting Ltd. Clustering libraries over the Aeron transport