GithubStar/UA-.NETStandard
1
0
Fork 0
mirror of https://github.com/OPCFoundation/UA-.NETStandard.git synced 2026-10-22 09:37:12 +08:00
Code Issues Packages Projects Releases Wiki Activity

Stabilize TCP receive (#904)

Browse Source 
- Various issues found during server testing with ref server
- Take a different approach on the nesting issue
- Fix a duplicate buffer free issue 
- stabilize binary decoder
This commit is contained in:
Martin Regen
2020-02-19 13:34:55 +01:00
committed by GitHub
parent 9f015128f0
commit e606f3aa8c
9 changed files with 684 additions and 699 deletions
Download Patch File Download Diff File Expand all files Collapse all files
SampleApplications/SDK/Opc.Ua.Client/Session.cs
+3 -3
View File
@@ -355,7 +355,7 @@ namespace Opc.Ua.Client
bool isLocalHost = false;
if (endpoint.EndpointUrl.HostNameType == UriHostNameType.Dns)
{
if (dnsHostName.ToLowerInvariant() == "localhost")
if (String.Equals(dnsHostName, "localhost", StringComparison.InvariantCultureIgnoreCase))
{
isLocalHost = true;
}
@@ -382,8 +382,8 @@ namespace Opc.Ua.Client
for (int ii = 0; ii < domains.Count; ii++)
{
if (String.Compare(hostname, domains[ii], StringComparison.OrdinalIgnoreCase) == 0 ||
String.Compare(dnsHostName, domains[ii], StringComparison.OrdinalIgnoreCase) == 0)
if (String.Equals(hostname, domains[ii], StringComparison.OrdinalIgnoreCase) ||
String.Equals(dnsHostName, domains[ii], StringComparison.OrdinalIgnoreCase))
{
domainFound = true;
break;
SampleApplications/Workshop/Reference/Server/ReferenceNodeManager.cs
+2 -2
View File
@@ -1,4 +1,4 @@
/* ========================================================================
/* ========================================================================
* Copyright (c) 2005-2019 The OPC Foundation, Inc. All rights reserved.
*
* OPC Foundation MIT License 1.00
@@ -570,7 +570,7 @@ namespace Quickstarts.ReferenceServer
CreateAnalogItemVariable(analogArrayFolder, daAnalogArray + "LocalizedText", "LocalizedText", BuiltInType.LocalizedText, ValueRanks.OneDimension, new LocalizedText[] { new LocalizedText("en", "Hello World1"), new LocalizedText("en", "Hello World2"), new LocalizedText("en", "Hello World3"), new LocalizedText("en", "Hello World4"), new LocalizedText("en", "Hello World5"), new LocalizedText("en", "Hello World6"), new LocalizedText("en", "Hello World7"), new LocalizedText("en", "Hello World8"), new LocalizedText("en", "Hello World9"), new LocalizedText("en", "Hello World10") });
CreateAnalogItemVariable(analogArrayFolder, daAnalogArray + "NodeId", "NodeId", BuiltInType.NodeId, ValueRanks.OneDimension, new NodeId[] { new NodeId(Guid.NewGuid()), new NodeId(Guid.NewGuid()), new NodeId(Guid.NewGuid()), new NodeId(Guid.NewGuid()), new NodeId(Guid.NewGuid()), new NodeId(Guid.NewGuid()), new NodeId(Guid.NewGuid()), new NodeId(Guid.NewGuid()), new NodeId(Guid.NewGuid()), new NodeId(Guid.NewGuid()) });
CreateAnalogItemVariable(analogArrayFolder, daAnalogArray + "Number", "Number", BuiltInType.Number, ValueRanks.OneDimension, new Int16[] { 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 });
CreateAnalogItemVariable(analogArrayFolder, daAnalogArray + "QualifiedName", "QualifiedName", BuiltInType.QualifiedName, ValueRanks.OneDimension, new Int16[] { 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 });
CreateAnalogItemVariable(analogArrayFolder, daAnalogArray + "QualifiedName", "QualifiedName", BuiltInType.QualifiedName, ValueRanks.OneDimension, new QualifiedName[] { "q0", "q1", "q2", "q3", "q4", "q5", "q6", "q7", "q8", "q9"});
CreateAnalogItemVariable(analogArrayFolder, daAnalogArray + "SByte", "SByte", BuiltInType.SByte, ValueRanks.OneDimension, new SByte[] { 10, 20, 30, 40, 50, 60, 70, 80, 90 });
CreateAnalogItemVariable(analogArrayFolder, daAnalogArray + "String", "String", BuiltInType.String, ValueRanks.OneDimension, new String[] { "a00", "b10", "c20", "d30", "e40", "f50", "g60", "h70", "i80", "j90" });
CreateAnalogItemVariable(analogArrayFolder, daAnalogArray + "Time", "Time", DataTypeIds.Time, ValueRanks.OneDimension, new String[] { DateTime.MinValue.ToString(), DateTime.MaxValue.ToString(), DateTime.MinValue.ToString(), DateTime.MaxValue.ToString(), DateTime.MinValue.ToString(), DateTime.MaxValue.ToString(), DateTime.MinValue.ToString(), DateTime.MaxValue.ToString(), DateTime.MinValue.ToString(), DateTime.MaxValue.ToString() }, null);
Stack/Opc.Ua.Core/Stack/State/BaseVariableState.cs
+5 -1
View File
@@ -1930,7 +1930,11 @@ namespace Opc.Ua
return (StatusCode)(uint)value;
}
}
return StatusCodes.BadTypeMismatch;
// test for special case Null type
if (!(m_dataType.IsNullNodeId && value == null))
{
return StatusCodes.BadTypeMismatch;
}
}
value = ExtractValueFromVariant(context, value, true);
Stack/Opc.Ua.Core/Stack/Tcp/TcpMessageSocket.cs
+112 -121
View File
@@ -52,8 +52,7 @@ namespace Opc.Ua.Bindings
{
public TcpMessageSocketAsyncEventArgs()
{
m_args = new SocketAsyncEventArgs
{
m_args = new SocketAsyncEventArgs {
UserToken = this
};
}
@@ -68,11 +67,7 @@ namespace Opc.Ua.Bindings
}
#endregion
public object UserToken
{
get { return m_userToken; }
set { m_userToken = value; }
}
public object UserToken { get; set; }
public void SetBuffer(byte[] buffer, int offset, int count)
{
@@ -117,7 +112,6 @@ namespace Opc.Ua.Bindings
}
public SocketAsyncEventArgs m_args;
private object m_userToken;
private event EventHandler<IMessageSocketAsyncEventArgs> m_internalComplete;
}
@@ -140,11 +134,7 @@ namespace Opc.Ua.Bindings
}
#endregion
public object UserToken
{
get { return m_UserToken; }
set { m_UserToken = value; }
}
public object UserToken { get; set; }
public void SetBuffer(byte[] buffer, int offset, int count)
{
@@ -181,7 +171,6 @@ namespace Opc.Ua.Bindings
}
private SocketError m_socketError;
private object m_UserToken;
}
@@ -236,7 +225,8 @@ namespace Opc.Ua.Bindings
m_bufferManager = bufferManager;
m_receiveBufferSize = receiveBufferSize;
m_incomingMessageSize = -1;
m_ReadComplete = new EventHandler<SocketAsyncEventArgs>(OnReadComplete);
m_readComplete = new EventHandler<SocketAsyncEventArgs>(OnReadComplete);
m_readState = ReadState.Ready;
}
/// <summary>
@@ -263,7 +253,7 @@ namespace Opc.Ua.Bindings
m_bufferManager = bufferManager;
m_receiveBufferSize = receiveBufferSize;
m_incomingMessageSize = -1;
m_ReadComplete = new EventHandler<SocketAsyncEventArgs>(OnReadComplete);
m_readComplete = new EventHandler<SocketAsyncEventArgs>(OnReadComplete);
}
#endregion
@@ -293,18 +283,7 @@ namespace Opc.Ua.Bindings
/// Gets the socket handle.
/// </summary>
/// <value>The socket handle.</value>
public int Handle
{
get
{
if (m_socket != null)
{
return m_socket.GetHashCode();
}
return -1;
}
}
public int Handle => m_socket != null ? m_socket.GetHashCode() : -1;
/// <summary>
/// Connects to an endpoint.
@@ -327,7 +306,7 @@ namespace Opc.Ua.Bindings
try
{
// Get DNS host information
hostAdresses = await Dns.GetHostAddressesAsync(endpointUrl.DnsSafeHost);
hostAdresses = await Dns.GetHostAddressesAsync(endpointUrl.DnsSafeHost).ConfigureAwait(false);
}
catch (SocketException e)
{
@@ -397,12 +376,12 @@ namespace Opc.Ua.Bindings
if (moreAddresses && !m_tcs.Task.IsCompleted)
{
await Task.Delay(1000);
await Task.Delay(1000).ConfigureAwait(false);
}
if (!moreAddresses || m_tcs.Task.IsCompleted)
{
error = await m_tcs.Task;
error = await m_tcs.Task.ConfigureAwait(false);
switch (error)
{
case SocketError.Success:
@@ -415,7 +394,7 @@ namespace Opc.Ua.Bindings
}
} while (moreAddresses);
ErrorExit:
ErrorExit:
doCallback(error);
return false;
@@ -459,30 +438,27 @@ namespace Opc.Ua.Bindings
/// Starts reading messages from the socket.
/// </summary>
public void ReadNextMessage()
{
var args = InternalReadNextMessage();
if (args != null)
{
m_ReadComplete(null, args);
}
}
private SocketAsyncEventArgs InternalReadNextMessage()
{
lock (m_readLock)
{
// allocate a buffer large enough to a message chunk.
if (m_receiveBuffer == null)
do
{
m_receiveBuffer = m_bufferManager.TakeBuffer(m_receiveBufferSize, "ReadNextMessage");
}
// allocate a buffer large enough to a message chunk.
if (m_receiveBuffer == null)
{
m_receiveBuffer = m_bufferManager.TakeBuffer(m_receiveBufferSize, "ReadNextMessage");
}
// read the first 8 bytes of the message which contains the message size.
m_bytesReceived = 0;
m_bytesToReceive = TcpMessageLimits.MessageTypeAndSize;
m_incomingMessageSize = -1;
// read the first 8 bytes of the message which contains the message size.
m_bytesReceived = 0;
m_bytesToReceive = TcpMessageLimits.MessageTypeAndSize;
m_incomingMessageSize = -1;
return ReadNextBlock();
do
{
ReadNextBlock();
} while (m_readState == ReadState.ReadNextBlock);
} while (m_readState == ReadState.ReadNextMessage);
}
}
@@ -508,50 +484,13 @@ namespace Opc.Ua.Bindings
try
{
bool innerCall = m_readState == ReadState.ReadComplete;
error = DoReadComplete(e);
while (ServiceResult.IsGood(error) && e != null)
// to avoid recursion, inner calls of OnReadComplete return
// after processing the ReadComplete and let the outer call handle it
if (!innerCall && !ServiceResult.IsBad(error))
{
e.Dispose();
// check if more data left to read.
if (m_bytesReceived < m_bytesToReceive)
{
e = ReadNextBlock();
}
// start reading the message body.
else if (m_incomingMessageSize < 0)
{
m_incomingMessageSize = BitConverter.ToInt32(m_receiveBuffer, 4);
if (m_incomingMessageSize <= 0 || m_incomingMessageSize > m_receiveBufferSize)
{
Utils.Trace(
"BadTcpMessageTooLarge: BufferSize={0}; MessageSize={1}",
m_receiveBufferSize,
m_incomingMessageSize);
error = ServiceResult.Create(
StatusCodes.BadTcpMessageTooLarge,
"Messages size {1} bytes is too large for buffer of size {0}.",
m_receiveBufferSize,
m_incomingMessageSize);
}
// set up buffer for reading the message body.
m_bytesToReceive = m_incomingMessageSize;
e = ReadNextBlock();
}
else
{
e = InternalReadNextMessage();
}
if(e != null)
{
error = DoReadComplete(e);
}
while (ReadNext()) ;
}
}
catch (Exception ex)
@@ -572,10 +511,7 @@ namespace Opc.Ua.Bindings
m_receiveBuffer = null;
}
if (m_sink != null)
{
m_sink.OnReceiveError(this, error);
}
m_sink?.OnReceiveError(this, error);
}
}
}
@@ -587,6 +523,7 @@ namespace Opc.Ua.Bindings
{
// complete operation.
int bytesRead = e.BytesTransferred;
m_readState = ReadState.Ready;
lock (m_socketLock)
{
@@ -606,6 +543,7 @@ namespace Opc.Ua.Bindings
m_receiveBuffer = null;
}
m_readState = ReadState.Error;
return ServiceResult.Create(StatusCodes.BadConnectionClosed, "Remote side closed connection");
}
@@ -614,12 +552,36 @@ namespace Opc.Ua.Bindings
// check if more data left to read.
if (m_bytesReceived < m_bytesToReceive)
{
m_readState = ReadState.ReadNextBlock;
return ServiceResult.Good;
}
// start reading the message body.
if (m_incomingMessageSize < 0)
{
m_incomingMessageSize = BitConverter.ToInt32(m_receiveBuffer, 4);
if (m_incomingMessageSize <= 0 || m_incomingMessageSize > m_receiveBufferSize)
{
Utils.Trace(
"BadTcpMessageTooLarge: BufferSize={0}; MessageSize={1}",
m_receiveBufferSize,
m_incomingMessageSize);
m_readState = ReadState.Error;
return ServiceResult.Create(
StatusCodes.BadTcpMessageTooLarge,
"Messages size {1} bytes is too large for buffer of size {0}.",
m_receiveBufferSize,
m_incomingMessageSize);
}
// set up buffer for reading the message body.
m_bytesToReceive = m_incomingMessageSize;
m_readState = ReadState.ReadNextBlock;
return ServiceResult.Good;
}
@@ -649,13 +611,16 @@ namespace Opc.Ua.Bindings
m_receiveBuffer = null;
}
// start receiving next message.
m_readState = ReadState.ReadNextMessage;
return ServiceResult.Good;
}
/// <summary>
/// Reads the next block of data from the socket.
/// </summary>
private SocketAsyncEventArgs ReadNextBlock()
private void ReadNextBlock()
{
Socket socket = null;
@@ -669,8 +634,8 @@ namespace Opc.Ua.Bindings
m_bufferManager.ReturnBuffer(m_receiveBuffer, "ReadNextBlock");
m_receiveBuffer = null;
}
return null;
m_readState = ReadState.NotConnected;
return;
}
socket = m_socket;
@@ -678,19 +643,19 @@ namespace Opc.Ua.Bindings
// avoid stale ServiceException when socket is disconnected
if (!socket.Connected)
{
return null;
m_readState = ReadState.NotConnected;
return;
}
}
BufferManager.LockBuffer(m_receiveBuffer);
ServiceResult error = ServiceResult.Good;
SocketAsyncEventArgs args = new SocketAsyncEventArgs();
var args = new SocketAsyncEventArgs();
try
{
m_readState = ReadState.Receive;
args.SetBuffer(m_receiveBuffer, m_bytesReceived, m_bytesToReceive - m_bytesReceived);
args.Completed += m_ReadComplete;
args.Completed += m_readComplete;
if (!socket.ReceiveAsync(args))
{
// I/O completed synchronously
@@ -698,28 +663,38 @@ namespace Opc.Ua.Bindings
{
throw ServiceResultException.Create(StatusCodes.BadTcpInternalError, args.SocketError.ToString());
}
else
{
return args;
//Root of the problem
//m_ReadComplete(null, args);
}
// set state to inner complete
m_readState = ReadState.ReadComplete;
m_readComplete(null, args);
}
}
catch (ServiceResultException)
{
args.Dispose();
args?.Dispose();
BufferManager.UnlockBuffer(m_receiveBuffer);
throw;
}
catch (Exception ex)
{
args.Dispose();
args?.Dispose();
BufferManager.UnlockBuffer(m_receiveBuffer);
throw ServiceResultException.Create(StatusCodes.BadTcpInternalError, ex, "BeginReceive failed.");
}
}
return null;
/// <summary>
/// Helper to read read next block or message based on current state.
/// </summary>
private bool ReadNext()
{
bool result = true;
switch (m_readState)
{
case ReadState.ReadNextBlock: ReadNextBlock(); break;
case ReadState.ReadNextMessage: ReadNextMessage(); break;
default: result = false; break;
}
return result;
}
/// <summary>
@@ -737,12 +712,11 @@ namespace Opc.Ua.Bindings
private SocketError BeginConnect(IPAddress address, AddressFamily addressFamily, int port, CallbackAction callback)
{
var socket = new Socket(addressFamily, SocketType.Stream, ProtocolType.Tcp);
var args = new SocketAsyncEventArgs()
{
var args = new SocketAsyncEventArgs() {
UserToken = callback,
RemoteEndPoint = new IPEndPoint(address, port),
};
args.Completed += new EventHandler<SocketAsyncEventArgs>(OnSocketConnected);
args.Completed += OnSocketConnected;
if (!socket.ConnectAsync(args))
{
// I/O completed synchronously
@@ -804,6 +778,7 @@ namespace Opc.Ua.Bindings
args.Dispose();
}
#endregion
#region Write Handling
/// <summary>
/// Sends a buffer.
@@ -823,29 +798,45 @@ namespace Opc.Ua.Bindings
return m_socket.SendAsync(eventArgs.m_args);
}
#endregion
#region Event factory
public IMessageSocketAsyncEventArgs MessageSocketEventArgs()
{
return new TcpMessageSocketAsyncEventArgs();
}
#endregion
#region Private Fields
private IMessageSink m_sink;
private BufferManager m_bufferManager;
private int m_receiveBufferSize;
private EventHandler<SocketAsyncEventArgs> m_ReadComplete;
private readonly int m_receiveBufferSize;
private readonly EventHandler<SocketAsyncEventArgs> m_readComplete;
private object m_socketLock = new object();
private readonly object m_socketLock = new object();
private Socket m_socket;
private bool m_closed = false;
private bool m_closed;
private TaskCompletionSource<SocketError> m_tcs;
private int m_socketResponses;
private object m_readLock = new object();
/// <summary>
/// States for the nested read handler.
/// </summary>
private enum ReadState
{
Ready = 0,
ReadNextMessage = 1,
ReadNextBlock = 2,
Receive = 3,
ReadComplete = 4,
NotConnected = 5,
Error = 0xff
};
private readonly object m_readLock = new object();
private byte[] m_receiveBuffer;
private int m_bytesReceived;
private int m_bytesToReceive;
private int m_incomingMessageSize;
private ReadState m_readState;
#endregion
}
}
Stack/Opc.Ua.Core/Stack/Tcp/TcpServerChannel.cs
+1 -1
View File
@@ -1208,7 +1208,7 @@ namespace Opc.Ua.Bindings
{
Utils.Trace(e, "Unexpected error processing request.");
SendServiceFault(token, requestId, ServiceResult.Create(e, StatusCodes.BadTcpInternalError, "Unexpected error processing request."));
return false;
return true;
}
finally
{
Stack/Opc.Ua.Core/Stack/Tcp/UaSCBinaryChannel.cs
+7 -7
View File
@@ -208,12 +208,8 @@ namespace Opc.Ua.Bindings
/// </summary>
protected void ChannelStateChanged(TcpChannelState state, ServiceResult reason)
{
Task.Run(() =>
{
if (m_StateChanged != null)
{
m_StateChanged(this, state, reason);
}
Task.Run(() => {
m_StateChanged?.Invoke(this, state, reason);
});
}
@@ -389,7 +385,6 @@ namespace Opc.Ua.Bindings
lock (DataLock)
{
ServiceResult error = ServiceResult.Good;
try
{
Utils.TraceDebug("Bytes written: {0}", e.BytesTransferred);
@@ -406,6 +401,11 @@ namespace Opc.Ua.Bindings
}
catch (Exception ex)
{
if (ex is InvalidOperationException)
{
// suppress chained exception in HandleWriteComplete/ReturnBuffer
e.BufferList = null;
}
error = ServiceResult.Create(ex, StatusCodes.BadTcpInternalError, "Unexpected error during write operation.");
HandleWriteComplete((BufferCollection)e.BufferList, e.UserToken, e.BytesTransferred, error);
}
Stack/Opc.Ua.Core/Types/Encoders/BinaryDecoder.cs
+1 -1
View File
@@ -1591,7 +1591,7 @@ namespace Opc.Ua
if (encodeable == null)
{
// figure out how long the object is.
if (length == -1)
if (length < 0)
{
throw new ServiceResultException(
StatusCodes.BadDecodingError,
Stack/Opc.Ua.Core/Types/Utils/DataGenerator.cs
+2 -1
View File
@@ -373,7 +373,8 @@ namespace Opc.Ua.Test
if (value != null)
{
if (expectedType == BuiltInType.Guid)
if (expectedType == BuiltInType.Guid &&
value is Guid)
{
value = new Uuid((Guid)value);
}
Stack/Opc.Ua.Core/Types/Utils/TypeInfo.cs
+551 -562
View File
File diff suppressed because it is too large Load Diff