Showing the changes to Source/Libraries/GSF.Core/Parsing/MultiSourceFrameImageParserBase.cs (1 of 147 files). Show all
Unified · Split
| @@ -37,7 +37,6 @@ using System; · 36 unmodified lines | ||
| 37 | 37 | using System.Collections.Concurrent; |
| 38 | 38 | using System.Collections.Generic; |
| 39 | 39 | using System.ComponentModel; |
| 40 | −using System.Threading; | |
| 41 | 40 | using GSF.Collections; |
| 42 | 41 | |
| 43 | 42 | namespace GSF.Parsing |
| @@ -54,7 +53,7 @@ namespace GSF.Parsing · 10 unmodified lines | ||
| 54 | 53 | /// This class is more specific than the <see cref="BinaryImageParserBase"/> in that it can automate the parsing of a |
| 55 | 54 | /// particular protocol that is formatted as a series of frames that have a common method of identification. |
| 56 | 55 | /// Automation of type creation occurs by loading implementations of common types that implement the |
| 57 | − /// <see cref="ISupportSourceIdentifiableFrameImage{TSourceIdentifier,TTypeIdentifier}"/> interface. The common method of | |
| 56 | + /// <see cref="ISupportSourceIdentifiableFrameImage{TSourceIdentifier, TTypeIdentifier}"/> interface. The common method of | |
| 58 | 57 | /// identification is handled by creating a class derived from the <see cref="ICommonHeader{TTypeIdentifier}"/> which primarily |
| 59 | 58 | /// includes a TypeID property, but also should include any state information needed to parse a particular frame if necessary. |
| 60 | 59 | /// Derived classes override the <see cref="FrameImageParserBase{TTypeIdentifier, TOutputType}.ParseCommonHeader"/> |
| @@ -334,8 +333,10 @@ namespace GSF.Parsing · 273 unmodified lines | ||
| 334 | 333 | // Copy buffer data for processing (destination buffer pre-allocated from buffer pool) |
| 335 | 334 | Buffer.BlockCopy(buffer, offset, identifiableBuffer.Buffer, 0, count); |
| 336 | 335 | |
| 337 | − // Add buffer to the queue for parsing | |
| 338 | − m_bufferQueue.Enqueue(new SourceIdentifiableBuffer[] { identifiableBuffer }); | |
| 336 | + // Add buffer to the queue for parsing. Note that buffer is queued for parsing instead | |
| 337 | + // of handling parse on this thread - this has become necessary to reduce UDP data loss | |
| 338 | + // that can happen in-process when system has UDP buffers building up for processing. | |
| 339 | + m_bufferQueue.Enqueue(new[] { identifiableBuffer }); | |
| 339 | 340 | } |
| 340 | 341 | } |
| 341 | 342 | catch |
| @@ -398,7 +399,7 @@ namespace GSF.Parsing · 56 unmodified lines | ||
| 398 | 399 | throw new NotImplementedException("This method should not be called directly, call the Parse(TSourceIdentifier,byte[],int,int) method to queue data for parsing instead."); |
| 399 | 400 | } |
| 400 | 401 | |
| 401 | − // This method is used by the internal <see cref="ProcessQueue{T}"/> to process all queued data buffers. | |
| 402 | + // This method is used to process all queued data buffers. | |
| 402 | 403 | private void ParseQueuedBuffers(IList<SourceIdentifiableBuffer> buffers) |
| 403 | 404 | { |
| 404 | 405 | if (Enabled) |
| @@ -427,6 +428,16 @@ namespace GSF.Parsing · 22 unmodified lines | ||
| 427 | 428 | } |
| 428 | 429 | finally |
| 429 | 430 | { |
| 431 | + // Dispose of source buffers - this returns buffers to pool | |
| 432 | + if ((object)buffers != null) | |
| 433 | + { | |
| 434 | + foreach (SourceIdentifiableBuffer buffer in buffers) | |
| 435 | + { | |
| 436 | + if ((object)buffer != null) | |
| 437 | + buffer.Dispose(); | |
| 438 | + } | |
| 439 | + } | |
| 440 | + | |
| 430 | 441 | // If user has attached to SourceDataParsed event, expose list of parsed data per source |
| 431 | 442 | if ((object)SourceDataParsed != null) |
| 432 | 443 | { |
| @@ -435,28 +446,10 @@ namespace GSF.Parsing · 2 unmodified lines | ||
| 435 | 446 | OnSourceDataParsed(parsedData.Key, parsedData.Value); |
| 436 | 447 | } |
| 437 | 448 | |
| 438 | − // Clear parsed data dictionary for next pass | |
| 449 | + // Clear parsed data dictionary for next pass - note that this does not contend with | |
| 450 | + // OnDataParsed override since the event called synchronously via base.Write above. | |
| 439 | 451 | m_parsedSourceData.Clear(); |
| 440 | 452 | } |
| 441 | − | |
| 442 | − } | |
| 443 | − } | |
| 444 | − | |
| 445 | − // Dispose of source buffers - no rush | |
| 446 | − ThreadPool.QueueUserWorkItem(DisposeSourceBuffers, buffers); | |
| 447 | − } | |
| 448 | − | |
| 449 | − // Handle disposing of source buffers (this will return items to the reusable object pool) | |
| 450 | − private void DisposeSourceBuffers(object state) | |
| 451 | − { | |
| 452 | − SourceIdentifiableBuffer[] buffers = state as SourceIdentifiableBuffer[]; | |
| 453 | − | |
| 454 | − if ((object)buffers != null) | |
| 455 | − { | |
| 456 | − foreach (SourceIdentifiableBuffer buffer in buffers) | |
| 457 | − { | |
| 458 | − if ((object)buffer != null) | |
| 459 | − buffer.Dispose(); | |
| 460 | 453 | } |
| 461 | 454 | } |
| 462 | 455 | } |