How to send large amount of Images DicomObjects
Common use case and restrictions
It is common that users need to send an entire DICOM study that contains hundreds (or even thousands) of images to a remote DICOM node. If using a single DICOM connection, the nature of DICOM C-STORE operation means that sending will be sequential as you cannot send the next image until you get response from the previous C-STORE. As a result, the transmission speed is not optimal and DICOM sending can be seen as “slow”.
To make matter worse, it is also possible that transcoding is happening when sending each DICOM image.
So to achieve the best transmission performance, it is important that sender knowns in advance the following 3 key factors:
- The SOP CLASS of the images
- The “storage” DICOM Transfer Syntax of each image on disk
- Filepath of each Image to send
The goal is to utilise multiple DICOM connections to send smaller “batch” in parallel, without having to transcode whenver it is possible.
Sample implementation of batch sending using DicomObjects.NET version
private const int MAX_NUMBER_OF_CONNECTIONS = 5; // maximum number of concurrent DICOM connections to send in parallel
private const int MAX_NUMBER_OF_IMAGES_PER_BATCH = 20; // number of images per Batch
public class DCMFile
{
public string SOPClass { get; set; }
public string OriginalTransferSyntax { get; set; }
public string FileLocation { get; set; }
}
public class RemoteAE
{
public string IP { get; set; }
public int Port { get; set; }
public string CallingAET { get; set; }
public string CalledAET { get; set; }
}
private void BatchSending(List<DCMFile> InstancesToSend, RemoteAE remoteAE)
{
// break up images into 1 or more queues
Queue sendQueue = Queue.Synchronized(new Queue(CreateBatches(InstancesToSend, MAX_NUMBER_OF_IMAGES_PER_BATCH)));
// create a sender thread for each queue
Thread[] workerThreads = new Thread[Math.Min(MAX_NUMBER_OF_CONNECTIONS, sendQueue.Count)];
for (int i = 0; i < workerThreads.Length; i++)
{
workerThreads[i] = new Thread(new ThreadStart(() =>
{
try
{
while (sendQueue.Peek() != null) // Check for images to send in the queue
{
// Get batch to process
IEnumerable<DCMFile> batchInstances = (IEnumerable<DCMFile>)sendQueue.Dequeue();
//Send instances to PACS
using (DicomAssociation assoc = new DicomAssociation())
{
bool dbUpdateRequired = false;
try
{
TlsInitiator GetTLSStream = null; // Default is unsecured
// Setup secured sending if needed
// Adding C-ECHO
assoc.RequestedContexts.Add(SOPClasses.Verification, TransferSyntaxes.ExplicitVRLittleEndian);
var sopClassGrouped = batchInstances.GroupBy(x => x.SOPClass);
// Explicitly adding SOP Class and TS
foreach (var g in sopClassGrouped)
{
var compiledTSs = TSstoPropose(g.Select(y => y.OriginalTransferSyntax));
foreach (string ts in compiledTSs)
assoc.RequestedContexts.Add(g.Key, ts);
}
if (IsPACSReachable(assoc, remoteAE, GetTLSStream))
{
dbUpdateRequired = true;
foreach (var instance in batchInstances)
{
int status = 0;
string errorComment = string.Empty;
try
{
// No need to open connection here, happens in DICOM echo
if (!assoc.isOpen)
{
// Try opening connection again
if (!IsPACSReachable(assoc, remoteAE, GetTLSStream))
break;
}
// Use original TS by choosing a Matching PCID
var matchingPCID = assoc.AgreedContexts.FirstOrDefault(x => x.AbstractSyntax == instance.SOPClass && x.AcceptedTS == instance.OriginalTransferSyntax)?.ContextID;
if (matchingPCID != null)
assoc.PreferredPCID = matchingPCID.Value;
else
assoc.PreferredPCID = 0; // Default
using (var ds = new DicomDataSet(instance.FileLocation))
{
TranscodeAndSend(assoc, instance, ds, out status, out errorComment);
}
}
catch (FileNotFoundException ex)
{
status = 0xC001;
errorComment = ex.Message;
}
catch (Exception ex)
{
status = 0xC000;
errorComment = ex.Message;
}
}
}
}
catch
{
// Swallow exception and carry on
}
finally
{
assoc.Close();
}
}
}
}
catch
{
// Swallow queue peek exception and carry on
}
}))
{
IsBackground = true,
Name = $"SenderThread #{i}",
Priority = ThreadPriority.AboveNormal
};
workerThreads[i].Start();
}
// Await all background threads
foreach (var workerThread in workerThreads)
workerThread.Join(600000); // 10 minutes thread timeout
}
internal static IEnumerable<T>[] CreateBatches<T>(IEnumerable<T> source, int chunkSize)
{
// Smaller chunks if total items are less
if (source.Count() <= MAX_NUMBER_OF_CONNECTIONS)
chunkSize = 1; // Minimum 1 item per batch
else if (source.Count() < chunkSize \* MAX_NUMBER_OF_CONNECTIONS)
{
// Adjust chunk size
chunkSize = source.Count() / MAX_NUMBER_OF_CONNECTIONS;
}
return source
.Select((x, i) => new { Index = i, Value = x })
.GroupBy(x => x.Index / chunkSize)
.Select(x => x.Select(v => v.Value).ToList())
.ToArray();
}
internal bool IsPACSReachable(DicomAssociation assoc, RemoteAE remoteAE, TlsInitiator GetTLSStream)
{
try
{
if (DICOMEcho(assoc, remoteAE, GetTLSStream) == 0)
return true;
else
{
return false; // Deem unreachable
}
}
catch
{
// Discard any exception and consider not reachable
return false;
}
}
internal int DICOMEcho(DicomAssociation assoc, RemoteAE remoteAE, TlsInitiator GetTLSStream)
{
assoc.InitiateTls += GetTLSStream;
assoc.Open(remoteAE.IP, remoteAE.Port, remoteAE.CallingAET, remoteAE.CalledAET);
return assoc.Echo();
}
/// <summary>
/// Propose only original transfer syntax of the images, and if necessary implicit vr little endian (mandatory)
/// This routine needs some extra attention, as you might want to experiment what gives you the best performance
/// say original transfer syntax is .90, PACS does not support .90, you will have 2 options:
/// A) decompress .90 to implicit vr little endian and send - uncompressed large data over the network
/// B) decompress .90 and recompress to .70 before send - extra compression but smaller data over the network
/// </summary>
/// <param name="originalTSs">the list of original transfer syntax of the images</param>
/// <returns>a list of transfer syntaxes to propose to remote PACS endpoint</returns>
internal IEnumerable<string> TSstoPropose(IEnumerable<string> originalTSs)
{
try
{
// Add Original TSs first
List<string> TSs = new List<string>(originalTSs);
// Making sure Default TS is included
TSs.Add(TransferSyntaxes.ImplicitVRLittleEndian);
TSs = TSs.Distinct().ToList(); // Avoid duplicate if OrignalTS is ImplicitVRLittleEndian
return TSs;
}
catch
{
// Swallow exception and rethrow
throw;
}
}
/// <summary>
/// this is the sending bit - we might have to transcode first to PACS endpoints selected transfer syntax
/// </summary>
internal static void TranscodeAndSend(DicomAssociation assoc, DCMFile instance, DicomDataSet ds, out int status, out string errorComment)
{
string agreedTS = assoc.AgreedContexts[assoc.PcidFor(instance.SOPClass)].AcceptedTS;
if (agreedTS != instance.OriginalTransferSyntax)
{
// Transcoding coding necessary so make sure that doesn't fail!
try
{
using (MemoryStream ms = new MemoryStream())
{
// Do stream write to ensure transcoding works
ds.Write(ms, agreedTS);
ms.Position = 0;
assoc.SendInstances(ms);
}
}
catch (Exception ex)
{
// failing to transcode, handle the exception here
}
}
else
assoc.SendInstances(ds); // Sending as-is, no transcoding
status = assoc.LastStatus;
errorComment = assoc.LastError;
}