Merge pull request #123 from SMI/release/v1.4.3

Release v1.4.3

CHANGELOG.md

@@ -9,6 +9,12 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 
 ...
 
+## [1.4.3] - 2020-02-21
+
+### Changed
+
+- Batch up RabbitMQ messages/acks in IdentifierMapper to avoid contention with the message publishing persistence
+
 ## [1.4.2] - 2020-02-18
 
 ### Added
@@ -170,7 +176,8 @@ First stable release after importing the repository from the private [SMIPlugin]
 - Anonymous `MappingTableName` must now be fully specified to pass validation (e.g. `mydb.mytbl`).  Previously skipping database portion was supported.
 
 
-[Unreleased]: https://github.com/SMI/SmiServices/compare/v1.4.2...develop
+[Unreleased]: https://github.com/SMI/SmiServices/compare/v1.4.3...develop
+[1.4.3]:  https://github.com/SMI/SmiServices/compare/v1.4.2...v1.4.3
 [1.4.2]:  https://github.com/SMI/SmiServices/compare/v1.4.1...v1.4.2
 [1.4.1]:  https://github.com/SMI/SmiServices/compare/v1.4.0...v1.4.1
 [1.4.0]:  https://github.com/SMI/SmiServices/compare/v1.3.1...v1.4.0

PACKAGES.md

@@ -15,7 +15,7 @@
 | HIC.RDMP.Dicom | [GitHub](https://github.com/HicServices/RdmpDicom) | [2.0.7](https://www.nuget.org/packages/HIC.RDMP.Dicom/2.0.7) | [GPL 3.0](https://www.gnu.org/licenses/gpl-3.0.html) | RDMP Plugin containing data load / pipeline components for imaging, reading dicom files etc | |
 | HIC.RDMP.Plugin | [GitHub](https://github.com/HicServices/RDMP) | [4.0.2](https://www.nuget.org/packages/HIC.RDMP.Plugin/4.0.2) | [GPL 3.0](https://www.gnu.org/licenses/gpl-3.0.html) | Interact with RDMP objects, base classes for plugin components etc | |
 | JetBrains.Annotations | | [2019.1.3](https://www.nuget.org/packages/JetBrains.Annotations/2019.1.3) |[MIT](https://opensource.org/licenses/MIT) | Static analysis tool | |
-| Microsoft.Extensions.Caching.Memory | [GitHub]() | [3.1.1](https://www.nuget.org/packages/Microsoft.Extensions.Caching.Memory/3.1.1) | | | |
+| Microsoft.Extensions.Caching.Memory | [GitHub](https://github.com/dotnet/extensions) | [3.1.2](https://www.nuget.org/packages/Microsoft.Extensions.Caching.Memory/3.1.2) | [Apache 2.0](https://www.nuget.org/packages/Microsoft.Extensions.Caching.Memory/3.1.2/License) | Caching ID mappings retrieved from Redis/MySQL |
 | MongoDB.Driver | [GitHub](https://github.com/mongodb/mongo-csharp-driver) |[2.9.3](https://www.nuget.org/packages/MongoDB.Driver/2.9.3)|  [Apache 2.0](https://www.nuget.org/packages/MongoDB.Driver/2.8.1/License) | For writting/reading dicom tags into MongoDb databases|
 | NLog | [GitHub](https://github.com/NLog/NLog) | [4.6.4](https://www.nuget.org/packages/NLog/4.6.4) | [BSD 3-Clause](https://github.com/NLog/NLog/blob/dev/LICENSE.txt) | Flexible user configurable logging | |
 | Newtonsoft.Json | [GitHub](https://github.com/JamesNK/Newtonsoft.Json) | [12.0.3](https://www.nuget.org/packages/Newtonsoft.Json/12.0.3) | [MIT](https://opensource.org/licenses/MIT) | Serialization of objects for sharing/transmission |

README.md

@@ -3,7 +3,7 @@
 ![GitHub](https://img.shields.io/github/license/SMI/SmiServices)
 [![Total alerts](https://img.shields.io/lgtm/alerts/g/SMI/SmiServices.svg?logo=lgtm&logoWidth=18)](https://lgtm.com/projects/g/SMI/SmiServices/alerts/)
 
-Version: `1.4.2`
+Version: `1.4.3`
 
 # SMI Services
 

docs/control-queues.md

@@ -14,7 +14,7 @@ This describes how the services can be controlled via RabbitMQ messages.
 
 Commands are sent by publishing a message to the ControlExchange (specified in your config by `RabbitOptions.RabbitMqControlExchangeName`) with a specific routing key. This allows you to easily send them from the RabbitMQ web management page, or via a CLI.
 
-RabbitMQ message routing keys are used to control which services recieve the message. The current format for routing keys is `smi.control.<who>.<what>`. Where `<who>` is the name of the service, and `<what>` is some defined action. The currently defined actions are:
+RabbitMQ message routing keys are used to control which services recieve the message. The current format for routing keys is `smi.control.<who>.<what>`. Where `<who>` is the name of the service, and `<what>` is some defined action. Note that all keys must be specified in lowercase. The currently defined actions are:
 
 ### General - any service
 

src/SharedAssemblyInfo.cs

@@ -7,6 +7,6 @@
 [assembly: AssemblyCulture("")]
 
 // These should be overwritten by release builds
-[assembly: AssemblyVersion("1.4.2")]
-[assembly: AssemblyFileVersion("1.4.2")]
-[assembly: AssemblyInformationalVersion("1.4.2")] // This one can have the extra build info after it
+[assembly: AssemblyVersion("1.4.3")]
+[assembly: AssemblyFileVersion("1.4.3")]
+[assembly: AssemblyInformationalVersion("1.4.3")] // This one can have the extra build info after it

src/common/Smi.Common/Execution/MicroserviceHost.cs

@@ -44,7 +44,7 @@ public abstract class MicroserviceHost : IMicroserviceHost
         /// </summary>
         /// <param name="globals">Settings for the microservice (location of rabbit, queue names etc)</param>
         /// <param name="loadSmiLogConfig">True to replace any existing <see cref="LogManager.Configuration"/> with the SMI logging configuration (which must exist in the file "Smi.NLog.config" of the current directory)</param>
-        protected MicroserviceHost(GlobalOptions globals, bool loadSmiLogConfig = true)
+        protected MicroserviceHost(GlobalOptions globals, bool loadSmiLogConfig = true, bool threaded=false)
         {
             HostProcessName = Assembly.GetEntryAssembly()?.GetName().Name ?? throw new ApplicationException("Couldn't get the Assembly name!");
 
@@ -112,8 +112,6 @@ protected MicroserviceHost(GlobalOptions globals, bool loadSmiLogConfig = true)
 
             OnFatal += (sender, args) => Fatal(args.Message, args.Exception);
 
-            // TODO(rkm 2020-02-14) This is only a temporary fix
-            bool threaded = (HostProcessName == "IdentifierMapper");
             RabbitMqAdapter = new RabbitMqAdapter(globals.RabbitOptions, HostProcessName + HostProcessID, OnFatal, threaded);
 
             _controlMessageConsumer = new ControlMessageConsumer(this, globals.RabbitOptions, HostProcessName, HostProcessID);

src/microservices/Microservices.IdentifierMapper/Execution/IdentifierMapperHost.cs

@@ -30,7 +30,7 @@ public class IdentifierMapperHost : MicroserviceHost
 
 
         public IdentifierMapperHost(GlobalOptions options, ISwapIdentifiers swapper = null, bool loadSmiLogConfig = true)
-            : base(options, loadSmiLogConfig)
+            : base(options, loadSmiLogConfig, threaded: true)
         {
             _consumerOptions = options.IdentifierMapperOptions;
 
@@ -66,10 +66,8 @@ public IdentifierMapperHost(GlobalOptions options, ISwapIdentifiers swapper = nu
                 throw;
             }
 
-            //TODO Probably want to run this in one of two modes:
-            //TODO 1) "Batch" -> Preload whole mapping table, process messages in batches (with batch consumer). Can't scale this horizontally (more services running)
-            //TODO 2) "Stream" -> Query the database for a swap value for each message as it comes in (with a small cache), produce single messages
-            _producerModel = RabbitMqAdapter.SetupProducer(options.IdentifierMapperOptions.AnonImagesProducerOptions, isBatch: false);
+            // Batching now handled implicitly as backlog demands
+            _producerModel = RabbitMqAdapter.SetupProducer(options.IdentifierMapperOptions.AnonImagesProducerOptions, isBatch: true);
 
             Consumer = new IdentifierMapperQueueConsumer(_producerModel, _swapper)
             {

src/microservices/Microservices.IdentifierMapper/Messaging/IdentifierMapperQueueConsumer.cs

@@ -1,11 +1,12 @@
-
 using Dicom;
 using DicomTypeTranslation;
 using Microservices.IdentifierMapper.Execution.Swappers;
 using RabbitMQ.Client.Events;
 using Smi.Common.Messages;
 using Smi.Common.Messaging;
 using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
 using System.Text.RegularExpressions;
 
 namespace Microservices.IdentifierMapper.Messaging
@@ -19,6 +20,7 @@ public class IdentifierMapperQueueConsumer : Consumer
 
         private readonly Regex _patientIdRegex = new Regex("\"00100020\":{\"vr\":\"LO\",\"Value\":\\[\"(\\d*)\"]", RegexOptions.IgnoreCase);
 
+        ConcurrentQueue<Tuple<DicomFileMessage,IMessageHeader,BasicDeliverEventArgs>> msgq=new ConcurrentQueue<Tuple<DicomFileMessage,IMessageHeader, BasicDeliverEventArgs>>();
 
         public IdentifierMapperQueueConsumer(IProducerModel producer, ISwapIdentifiers swapper)
         {
@@ -70,11 +72,33 @@ protected override void ProcessMessageImpl(IMessageHeader header, BasicDeliverEv
             else
             {
                 // Now ship it to the exchange
+                // Looks strange, but:
+                // First invocation will send 1 message and wait for ack
+                // While that is waiting, more messages pile up
+                // Once that completes, it releases the lock and another thread will drain the whole queue in one
+                // This should then have the effect of batching messages up and waiting for a full batch to process.
+                msgq.Enqueue(new Tuple<DicomFileMessage, IMessageHeader, BasicDeliverEventArgs>(msg, header, deliverArgs));
                 lock (_producer)
                 {
-                    _producer.SendMessage(msg, header);
+                    if (!msgq.IsEmpty)
+                    {
+                        List<Tuple<IMessageHeader, BasicDeliverEventArgs>> done = new List<Tuple<IMessageHeader, BasicDeliverEventArgs>>();
+                        while (!msgq.IsEmpty)
+                        {
+                            Tuple<DicomFileMessage, IMessageHeader, BasicDeliverEventArgs> t;
+                            if (msgq.TryDequeue(out t))
+                            {
+                                _producer.SendMessage(t.Item1, t.Item2, "");
+                                done.Add(new Tuple<IMessageHeader, BasicDeliverEventArgs>(t.Item2, t.Item3));
+                            }
+                        }
+                        _producer.WaitForConfirms();
+                        foreach (var t in done)
+                        {
+                            Ack(t.Item1, t.Item2);
+                        }
+                    }
                 }
-                Ack(header, deliverArgs);
             }
         }
 
@@ -155,4 +179,4 @@ public bool SwapIdentifier(DicomFileMessage msg, out string reason)
             return true;
         }
     }
-}
+}

src/microservices/Microservices.IdentifierMapper/Microservices.IdentifierMapper.csproj

@@ -16,7 +16,7 @@
   </ItemGroup>
   <ItemGroup>
     <PackageReference Include="StackExchange.Redis" Version="2.0.601" />
-    <PackageReference Include="Microsoft.Extensions.Caching.Memory" Version="3.1.1" />
+    <PackageReference Include="Microsoft.Extensions.Caching.Memory" Version="3.1.2" />
   </ItemGroup>
   <ItemGroup>
     <ProjectReference Include="..\..\common\Smi.Common\Smi.Common.csproj" />