Synchronizing SimCorp Dimension Trades
A frequent issue in SimCorp Dimension, is when data needs to be imported in a specific order. The scenario given in this post is of an initial trade import, followed by subsequent trade updates to the initial trade. The issue arising when messages or files from upstream systems are misaligned and updates arrive before the initial trade import, making the trade update fail in SimCorp Dimension and a support nightmare unfolds trying to untangle the failing trades.
With the SimCorp Dimension Communication Server comes great flexibility in how data is handled, both in terms of scalability (simply start more of the same server) and the asynchronous nature of the Communication Server (designed correctly the multithreaded nature of the Server offers the possibility of few bottlenecks).
So how could we ensure that a trade is imported into SimCorp Dimension before subsequent updates are executed against the database? Well, one obvious answer could be to look up the data in the Dimension database either directly from the Communication Server or through a Data Extractor call to see if the trade is present. This would however be somewhat excessive, when taking into account that this scenario occurs relatively infrequently and would cause unnecessary database activity in a production environment.
The example demonstrated in this post offers a different solution using the persistence and critical section functionality in the Communication Server. Persistence to keep track of trades (even in a multi-server environment) and the critical section to synchronize code in a multithreaded environment (this again also applies to a multi-server environment).
The scenario given in this post uses files to represent trade data with one directory for new trades and another for updates. The code works off a common trade Id, which could be the original transaction number in Dimension. The example does not go into detail with how trades are imported into SimCorp Dimension, but instead focuses on the synchronization mechanism explored in this post. How to import trades and other data into Dimension I will leave for another blog post.
The file for the example really only has one important column and that is the Trade Id which I have placed in a tab delimited text file in the first column.
| Trade Id | Column 2 | Column 3 |
|---|---|---|
| 123456789 | ABC Security | 1500 |
There are also a few constants set up in the config file and referenced in the cartridge file.
| Constant | Description |
|---|---|
| C_SEQTRADE | Name of the critical section |
| PF_SEQTRADE | Prefix of the critical section. The prefix and name will ensure a unique name, irrespective of which server solution the name is used in. |
| CS_TIMEOUT | The critical section can use a timeout section if necessary. This example does not deal with timeout and has this value set to a high value of 3600 seconds. |
| ID_COLUMN | The column in the file which holds the Trade Id. This example uses the first column and has this constant set to 1. |
There are also a few ports set up in the config file, which are discussed below.
| Port | Type | Description |
|---|---|---|
| createTrade | File | Files for incoming trades to be created in SimCorp Dimension. |
| updateTrade | File | Files for incoming trades to be updated in SimCorp Dimension. |
| processUpdate | Dimension | Port for handling trade updates in SimCorp Dimension. |
| tradeCreated | Persistence | Persistence port for trades that have been created. |
| saveTrade | Persistence | Persistence port for trades that have not yet been created in SimCorp Dimension and are ready to update. |
| sequenceTimer | Timer | Timer port to handle trades stored on the saveTrade persistence port. |
There are four blocks of code in this example. The first handles new trades from the createTrade port and creates them in SimCorp Dimension. The second handles updated trades from the updateTrade port. The third investigates whether trades have been created before updating them. Finally the fourth block of code is connected to the timer port and handles all trades that are yet to be updated.
In the first block of code, there are three tasks to attend to. Extract the tab delimited data, create the trades in SimCorp Dimension and flag the trade as created by persisting the trade id.
<inputport port="createTrade"> <!-- Create new trades --> <handler label="(Default)" output="file"> <try> <do> <!-- Extract values from tab file --> <extractdata input="file" output="dataSet" delimiter="	" recorddelimiter="
" /> <!-- Loop through the trade file --> <foreach input="dataSet" elementpath="/Data/Record" output="dataRecord"> <criticalsection name="@const(C_SEQTRADE)" prefix="@const(PF_SEQTRADE)" timeout="@const(CS_TIMEOUT)"> <lock> <!-- Pick out the trade ID --> <getvalue input="dataRecord" conditionpath="/Record/Column[@const(ID_COLUMN)]/text()" output="dataId" /> <!-- Create trade in SimCorp Dimension now --> <!-- ... --> <!-- Flag trade as created --> <persistsave input="dataId" port="tradeCreated" id="@var(dataId)" /> </lock> </criticalsection> </foreach> </do> <catch exception="exception" details="details"> <notify message="***Error: @msg(port): @msg(label) - @var(exception) - @var(details)" /> </catch> </try> </handler> </inputport>
In the code above, the creation of the trade and the trade id are executed as one block, uninterrupted by any other threads investigating the same trade id. The code has been synchronized through the critical section.
In the second block of code, there are two tasks to attend to. Extract the tab delimited data and send the individual trades on to a dedicated handler to process the trade update.
<inputport port="updateTrade"> <!-- Update trades --> <handler label="(Default)" output="file"> <try> <do> <!-- Extract values from tab file --> <extractdata input="file" output="dataSet" delimiter="	" recorddelimiter="
" /> <!-- Loop through the trade file --> <foreach input="dataSet" elementpath="/Data/Record" output="dataRecord"> <send input="dataRecord" label="-" port="processUpdate" /> </foreach> </do> <catch exception="exception" details="details"> <notify message="***Error: @msg(port): @msg(label) - @var(exception) - @var(details)" /> </catch> </try> </handler> </inputport>
The third block of code will investigate whether the trade has been created in SimCorp Dimension and then either update the trade or save the trade for later once the trade has been created in SimCorp Dimension. Like the first block of code where the trades are created, the logic that determines whether to update a trade or not is executed as an uninterrupted block of code. New trades will not be created or updated while this block is executing.
<inputport port="processUpdate"> <!-- Process update --> <handler label="(Default)" output="record"> <try> <do> <!-- Pick out the trade ID --> <getvalue input="record" conditionpath="/Record/Column[@const(ID_COLUMN)]/text()" output="dataId" /> <!-- Has this trade been created so we can update the trade --> <criticalsection name="@const(C_SEQTRADE)" prefix="@const(PF_SEQTRADE)" timeout="@const(CS_TIMEOUT)"> <lock> <!-- Find created trade --> <persistload port="tradeCreated" id="@var(dataId)" ignore="yes" remove="no" output="foundID" /> <ifelse conditionpath="'@var(dataId)' = '@var(foundID)'"> <then> <!-- Great, trade was created. Update trade in SimCorp Dimension now --> <!-- ... --> </then> <else> <!-- ID not created, persist trade for later --> <persistsave input="record" port="saveTrade" id="@var(dataId)" /> </else> </ifelse> </lock> </criticalsection> </do> <catch exception="exception" details="details"> <notify message="***Error: @msg(port): @msg(label) - @var(exception) - @var(details)" /> </catch> </try> </handler> </inputport>
In order to process the trades that are waiting to be updated (waiting for a trade to be created in Dimension first), a timer is used to run through the list. The entire list is read and processed as one block of code sending the trades to the above processUpdate port to be updated.
<inputport port="sequenceTimer"> <!-- Sequence trades --> <handler label="(Default)" output="tick"> <try> <do> <criticalsection name="@const(C_SEQTRADE)" prefix="@const(PF_SEQTRADE)" timeout="@const(CS_TIMEOUT)"> <lock> <!-- Retrieve saved trades --> <persistlist port="saveTrade" output="tradeList" /> <!-- Loop through the trade Id's --> <foreach input="tradeList" elementpath="/list/persist" output="persistItem"> <!-- Load trade to process --> <getvalue input="persistItem" conditionpath="string(/persist/@id)" output="dataId" /> <persistload port="saveTrade" id="@var(dataId)" ignore="no" remove="yes" output="record" /> <send input="record" label="-" port="processUpdate" /> </foreach> </lock> </criticalsection> </do> <catch exception="exception" details="details"> <notify message="***Error: @msg(port): @msg(label) - @var(exception) - @var(details)" /> </catch> </try> </handler> </inputport>
Of course, this example could be extended to become more elaborate. Should there be a retry counter for each update and should we query the database for the trade after a maximum number of retries? I leave that for you to investigate and hope I have succeeded in showing you how trade creation can be synchronized to occur before any trade updates in SimCorp Dimension using the Communication Server with persistence and critical sections.