# Data Relay and Persistent Event Queue Strategies and Implementation

> A transaction log that is usable for restoring the main eXtremeDB database (which is often the primary purpose of Transaction Logging) may or may not be created. Data to potentially be passed to an external data store is read by the function `mco_translog_iterate()` which calls in its turn a user-defined call-back function. The data `mco_translog_iterate()` passes to this call-back function are objects that were created, deleted or modified in the eXtremeDB database inside logged transactions. (Note that this is comparable to using the standard eXtremeDB function `mco_trans_iterate()` to iterate through the objects effected by a transaction, except that `mco_trans_iterate()` only iterates though the objects in the current transaction, and thus provides the capability of exporting this data synchronously, whereas with `mco_translog_iterate()` a log file can be read and “exported” at any time.)

> There are two ways to supply data from a writer process (the process that started transaction logging and is currently processing data inside the database) to a `reader` process (the process reading data and transferring it to the external data store by calling the user-defined callback).

> The first method is to create a log file. After the log file has been closed by the log writer, this log can be opened and read for export just as it could be opened to restore the main database as in ordinary TL usage.

> The second method is to use the pipe mechanism supplied by eXtremeDB Transaction Logging. In this case, data is being transferred through a memory buffer, either in conventional memory or in shared memory. This method is faster than the file-based method, but can create a race situation: if the `reader` process (which is calling the function `mco_translog_iterate()`) does not read data fast enough, then the `writer` process blocks with the result that there is no available memory in the pipe buffer. This causes the `writer` process to be blocked until the pipe buffer has available memory (that is, until the `reader` process releases the necessary amount of pipe buffer). This problem can be solved by extending the main pipe buffer with a temporary file in case of buffer overflow; then no blocking occurs inside the `writer` process.

> The two methods can be combined by setting the flag `MCO_TRANSLOG_DUAL_OUT`.

> Be aware that when using TL with hybrid databases, performance can be improved by turning off disk manager logging (passing the `NO_LOG` log type into the `mco_db_open_dev()` API). If a database contains only persistent classes, then the only reason to use TL is to export transactions to an external system because reXtremeDB has its own integral transaction logging implementation that is separate from the eXtremeDB Transaction Logging optional module.

> The asynchronous export is started by calling the function `mco_translog_start()` with the flag `MCO_TRANSLOG_ITERABLE`. This allows reading and then iterating the log by calling the function `mco_translog_iterate()`.

> Whether a file or pipe is used between the reader and writer processes the flag `MCO_TRANSLOG_ITERABLE` is necessary to start the relay process correctly. The `writer` and `reader` may be threads of a single process or they may be different processes (applications) running on a single computer. When using the file-based method, these applications may even be on different computers – if the process that is applying the logged transactions to the external data store is on a different computer, then the file-based method, though slower, is the only option.

> ## Ordinary log file as a transport to a reader

> To use an ordinary log file for exporting transactions to an external data store, the file must be opened by function `mco_translog_iterate()`. To be a valid log file, it has to have been closed by `mco_translog_stop()` on the `writer` side, or it has to have been indirectly closed by calling `mco_translog_start()` with `MCO_TRANSLOG_RESTART`. (Note that it is possible to use additional files as semaphores to indicate that a log file was properly closed by a writer.)

> #### Opening simple file based log:

>         #define DISK_PAGE_SIZE 4096
>         #define LOG_FILE       "/tmp/mylog"
>         mco_TL_start_data_t log_parms;
>         log_parms.flags = MCO_TRANSLOG_ITERABLE;
>         log_parms.disk_page_size = DISK_PAGE_SIZE;
>         mco_translog_start (db, LOG_FILE, &log_parms);

> #### Reading and iteration of log file:

>         #define TEMP_BUFFER_SIZE (32*1024*1024)
>         #define LOG_FILE       "/tmp/mylog"
>         MCO_RET iteration_proc (mco_trans_h trans, MCO_Hf* obj, int cid, int obj_state, void* user_ctx)
>         {
>             return MCO_S_OK;
>         }
>         ...
>         mem_buff = malloc (TEMP_BUFFER_SIZE);
>         mco_translog_iterate (LOG_FILE, 0, iteration_proc, user_ctx, mydatabase_get_dictionary (),

>                         mem_buff, TEMP_BUFFER_SIZE);

> ### Partitioning the log into multiple files

> To reduce the delay between the moment when a transaction is written into the log and when it is passed to the user-defined iteration callback, and also to reduce the disk space used by log files, it is possible to limit the maximum size of a log file and switch logging to another file when that limit has been reached.

> There are two ways to implement this file size limit. The first method is by starting the logging with flag `MCO_TRANSLOG_SIZE_CLBK`, a maximum log file size and the address of a user-defined file-size callback function. This implements a kind of `interrupt` strategy; when the specified log file size is exceeded the callback function is called.

> The second method is a `polling` strategy. The current log file size can be periodically checked by function `mco_translog_get_info()`.

> With both of these methods the log file size is being checked from the `writer` process.

> When the log file needs to be switched to another, simply call the function `mco_translog_start()` again but with the additional flag `MCO_TRANSLOG_RESTART`. This causes the log file to be switched to another without the necessity of calling the function `mco_translog_stop()`, and also without having to freeze transaction activity for this period, which is very advantageous for multithreaded applications.

> #### Example of writing a multiple file log – method 1: “interrupt”

>         #define DISK_PAGE_SIZE 4096
>         #define MAX_LOG_SIZE   128*1024*1024
>         #define LOG_FILE1       "/tmp/mylog1"
>         #define MARK_FILE1      "/tmp/mymark1"
>         #define LOG_FILE2       "/tmp/mylog2"
>         #define MARK_FILE2      "/tmp/mymark2"
>         volatile int            size_exceeded = 0;

>         void warn_sz_proc (mco_size_t log_size)
>         {
>             size_exceeded = 1;
>         }

>         ...

>         mco_TL_current_info_t log_info;
>         mco_TL_start_data_t log_parms;
>         log_parms.flags = MCO_TRANSLOG_ITERABLE | MCO_TRANSLOG_SIZE_CLBK;
>         log_parms.disk_page_size = DISK_PAGE_SIZE;
>         log_parms.max_size = MAX_LOG_SIZE;
>         log_parms.warn_sz_proc = warn_sz_proc;
>         mco_translog_start (db, LOG_FILE1, &log_parms);

>         /* Work with database. Current log is the file LOG_FILE1 */
>         while (!size_exceeded)
>         {
>             ...
>         }
>         log_parms.flags |= MCO_TRANSLOG_RESTART;
>         mco_translog_start (db, LOG_FILE2, &log_parms);
>         create_file (MARK_FILE1);

>         /* Work with database further. Current log is the file LOG_FILE2 */
>         ...
>         mco_translog_stop (db);
>         create_file (MARK_FILE2);

> #### Example of writing a multiple file log – method 2: “polling”

>         #define DISK_PAGE_SIZE 4096
>         #define MAX_LOG_SIZE   128*1024*1024
>         #define LOG_FILE1       "/tmp/mylog1"
>         #define MARK_FILE1      "/tmp/mymark1"
>         #define LOG_FILE2       "/tmp/mylog2"
>         #define MARK_FILE2      "/tmp/mymark2"

>         mco_TL_current_info_t log_info;
>         mco_TL_start_data_t log_parms;
>         log_parms.flags = MCO_TRANSLOG_ITERABLE;
>         log_parms.disk_page_size = DISK_PAGE_SIZE;
>         mco_translog_start (db, LOG_FILE1, &log_parms);

>         /* Work with database. Current log is the file LOG_FILE1 */
>         while (mco_translog_get_info (db, &log_info) == MCO_S_OK
>                 && log_info.log_size < MAX_LOG_SIZE)
>         {
>             ...
>         }
>         log_parms.flags |= MCO_TRANSLOG_RESTART;
>         mco_translog_start (db, LOG_FILE2, &log_parms);
>         create_file (MARK_FILE1);

>         /* Work with database further. Current log is the file LOG_FILE2 */
>         ...
>         mco_translog_stop (db);
>         create_file (MARK_FILE2);

> ### Reading and iteration of multiple log files

> Reading from multiple log files is quite simple. The application simply switches log files by sequentially calling function `mco_translog_iterate()`.

> #### Example of reading and iteration of multiple log files

>         #define TEMP_BUFFER_SIZE (32*1024*1024)
>         #define LOG_FILE1       "/tmp/mylog1"
>         #define MARK_FILE1      "/tmp/mymark1"
>         #define LOG_FILE2       "/tmp/mylog2"
>         #define MARK_FILE2      "/tmp/mymark2"
>         MCO_RET iteration_proc (mco_trans_h trans, MCO_Hf* obj, int cid, int obj_state, void* user_ctx)
>         {
>             /* Note "trans" is always 0 in case of asynchronous export */
>             return MCO_S_OK;
>         }
>         ...
>         mem_buff = malloc (TEMP_BUFFER_SIZE);

>         wait_while_no_file (MARK_FILE1);

>         mco_translog_iterate (LOG_FILE1, 0, iteration_proc, user_ctx, mydatabase_get_dictionary (),
>                         mem_buff, TEMP_BUFFER_SIZE);

>         unlink (MARK_FILE1);
>         wait_while_no_file (MARK_FILE2);

>         mco_translog_iterate (LOG_FILE2, 0, iteration_proc, user_ctx, mydatabase_get_dictionary (),
>                         mem_buff, TEMP_BUFFER_SIZE);
>         unlink (MARK_FILE2);

> Please note that the function `mco_translog_iterate()` returns only after it has completed reading of a log file or if the user defined call-back returns a value other than `MCO_S_OK` or if some other internal error occurs. Also note that it is necessary to call `mco_translog_iterate()` repeatedly to read the next log file.

> ## Using a pipe to transport data

> When pipe mode is used as a transport for exporting transaction data to an external database, a pipe-buffer is placed in shared or conventional memory (depending on the memory settings used in the main eXtremeDB database being used as the data source). If the `reader` process is processing data faster than the `writer` process is placing it into the pipe, then this mode is clearly the fastest. However, if the reader is slower than the writer process, the pipe buffer will overflow and the writer process will be blocked inside the transaction commit until the reader releases enough memory to store the transaction.

> It is possible to avoid this blocking by using an optional overflow file. This file is specified in the `file_path` argument of functions `mco_translog_start()` and `mco_translog_iterate()`. If a file is specified, it is used if pipe buffer overflow occurs and, therefore, the writer process is never blocked.

> Naturally, the `writer` and `reader` processing with this overflow file will be much slower. This is a consideration when determining the pipe buffer size. Another consideration is that, in any case, pipe mode does not generate log files that can be used for recovery of a database.

> The reader process must open a database handle (by calling `mco_db_connect()`) of the database for which some or all transactional data is to be exported. This database handle, in turn, must be passed to the function `mco_translog_iterate()`. This allows access to the pipe from the reader.

> The pipe buffer is an instance of type `mco_device_t` and must be specified when the database is created. It will be assigned `MCO_MEMORY_ASSIGN_PIPE_BUF` and type `MCO_MEMORY_NAMED` or `MCO_MEMORY_CONV`, which must be the same memory type as that used by the main database device.

> It is also possible to specify more than one pipe buffering device. In this case transactions will automatically be broadcast to all pipe instances as separate pipe buffers. This implies that the count of readers (processes or threads calling function `mco_translog_iterate()` or `mco_translog_play()`) must be equal to the count of pipe devices. It is useful to check the field `pipe_readers_connected` of the `mco_TL_current_info` structure to verify that all readers are connected. As all of the pipe devices work concurrently, it is recommended to specify the same size for all of the pipes. And likewise for all temporary buffers passed into `mco_translog_iterate()`. (See sample tlogitermultipipe for an example implementation of multiple pipes.)

> The temporary buffer passed into `mco_translog_iterate()` should be allocated by `malloc()` if the main database device is of type conventional memory. But if shared (named) memory is used by the main database device, then the pointer to a temporary buffer passed to `mco_translog_iterate()` must be 0. In this case, `mco_translog_iterate()` will create a temporary buffer in shared memory automatically.

> > Note that, when using shared memory, it is important to distinguish between the 'offset' and 'direct pointer' nature of the eXtremeDB core. If the 'dptr' core is used then all memory devices (objects of structure `mco_device_t`) must specify some unique 'hint' field which is actually the beginning address of the requested shared memory block. Regarding functions mco_translog_play(),and mco_translog_iterate() , the argument (or corresponding field of structure `mco_TL_play_params_h`) `void *mem_ptr` is actually related to an internally used memory device which temporarily holds database objects read from a log stream. Thus the argument `mem_ptr` here is the same 'hint' for that memory device. It should be zero in the case of the 'offset' core or should be a valid address value in the case of 'dptr' core.

> ### Pre-buffering

> When initializing the pipe, the application can specify the `MCO_TRANSLOG_SYNC_INSTANTLY` flag in the `data_flags` structure. If specified, then each transaction is written into the pipe immediately. Otherwise the Transaction Logging runtime buffers transactions in its internal intermediate buffer (the default size is `64K`). When the buffer becomes full it is written into the pipe all at once. The flag values `MCO_TRANSLOG_SYNC_COUNT` and `MCO_TRANSLOG_SYNC_TIMER` are not allowed when using multiple pipes and dual out mode (see below).

> The user-defined callback `iteration_proc` may report an error by returning a value different from `MCO_S_OK`. In this case `mco_translog_iterate()` returns immediately with the same return code that the `iteration_proc` has just returned. On the writer side the transaction commit (`mco_trans_commit()` or `mco_trans_commit_phase2()`) will return error code `MCO_E_TL_PIPE_TERM`. This informs the writer process that the reader has broken its loop. In this case, the transaction log should be stopped by calling `mco_translog_stop()` before any further actions.

> ### Option Mark_Last_Object

> The runtime flag `MCO_RT_OPTION_MARK_LAST_OBJ` can be set (by function `mco_runtime_setoption()`) to specify that notification of ”end-of-transaction” is required. Transactions in a log file or pipe stream are output in their original order. By default while reading and iterating the log or pipe there is no simple way to understand if the object passed into the user-defined iteration callback is the latest in the current transaction. For this purpose the runtime option `MCO_RT_OPTION_MARK_LAST_OBJ` is used to enable the runtime to notify the user-defined callback about the end of each transaction. If it is set then user-defined iteration callback will be called one additional time with a zeroed object handle to indicate that the current transaction has finished.

> Further implementation details are demonstrated in samples tlogiterpipe, tlogitermultipipe and tlogiterdualout.

> #### Opening a pipe-based log

>         #define LOG_PAGE_SIZE 128
>         #define PIPE_BUFFER_SIZE (16*1024*1024)
>         mco_device_t dev[];
>         mco_TL_start_data_t log_parms;
>         ...

>         /* assign pipe buffer as mco device. It will be automatically
>         passed to a reader */
>         dev[x].assignment = MCO_MEMORY_ASSIGN_PIPE_BUF;
>         dev[x].size = PIPE_BUFFER_SIZE;
>         dev[x].type = MCO_MEMORY_NAMED;

>         sprintf( dev[x].dev.named.name, "%s-pipe", databaseName_ );
>         dev[x].dev.named.flags = 0;
>         dev[x].dev.named.hint  = 0;
>         ...

>         mco_db_open_dev (DATABASE_NAME, mydatabase_get_dictionary(), dev, x, &db_params);
>         mco_db_connect (DATABASE_NAME, &db);
>         log_parms.flags = MCO_TRANSLOG_ITERABLE | MCO_TRANSLOG_PIPE;
>         log_parms.disk_page_size = LOG_PAGE_SIZE;

>         mco_translog_start (db, 0, &log_parms);

> #### Reading and iteration of a pipe-based log

>         #define TEMP_BUFFER_SIZE (32*1024*1024)
>         #define PIPE_BUFFER_SIZE (16*1024*1024)

>         MCO_RET iteration_proc (mco_trans_h trans, MCO_Hf* obj, int cid, int obj_state, void* user_ctx)
>         {
>             mco_trans_counter_t trans_no;
>             mco_trans_no (trans, &trans_no);
>             if (all_ok)
>             return MCO_S_OK;
>             else
>             return MCO_ERR_LAST + 1;
>         }
>         ...

>         MCO_RET rc;
>         void *mem_buff;

>         mco_db_connect (mydatabase_name, &db);

>         /* if used memory management library is different from mcomconv */
>         #ifdef USE_SHARED_MEMORY
>         mem_buff = 0;
>         #else
>         mem_buff = malloc (TEMP_BUFFER_SIZE);
>         #endif

>         mco_translog_iterate (0, db, iteration_proc, user_ctx, mydatabase_get_dictionary (), mem_buff, TEMP_BUFFER_SIZE);

> #### Opening a pipe-based log with overflow file

>         #define LOG_PAGE_SIZE 128
>         #define PIPE_BUFFER_SIZE (16*1024*1024)
>         #define PIPE_OVEFLOW_FILE "/tmp/overflow_file"
>         mco_device_t dev[];
>         mco_TL_start_data_t log_parms;
>         ...

>         /* assign pipe buffer as mco device. It will be automatically
>         passed to a reader */
>         dev[x].assignment = MCO_MEMORY_ASSIGN_PIPE_BUF;
>         dev[x].size = PIPE_BUFFER_SIZE;
>         dev[x].type = MCO_MEMORY_NAMED;
>         sprintf( dev[x].dev.named.name, "%s-pipe", databaseName_ );
>         dev[x].dev.named.flags = 0;
>         dev[x].dev.named.hint  = 0;

>         ...

>         mco_db_open_dev (DATABASE_NAME, mydatabase_get_dictionary(), dev, x, &db_params);

>         mco_db_connect (DATABASE_NAME, &db);

>         log_parms.flags = MCO_TRANSLOG_ITERABLE | MCO_TRANSLOG_PIPE;
>         log_parms.disk_page_size = LOG_PAGE_SIZE;

>         mco_translog_start (db, PIPE_OVEFLOW_FILE, &log_parms);

> #### Reading and iteration of a pipe-based log with overflow file

>         #define TEMP_BUFFER_SIZE (32*1024*1024)
>         #define PIPE_BUFFER_SIZE (16*1024*1024)
>         #define PIPE_OVEFLOW_FILE "/tmp/overflow_file"

>         MCO_RET iteration_proc (mco_trans_h trans, MCO_Hf* obj, int cid, int obj_state, void* user_ctx)
>         {
>             mco_trans_counter_t trans_no;
>             mco_trans_no (trans, &trans_no);
>             ...

>             if (all_ok)
>             return MCO_S_OK;
>             else
>             return MCO_ERR_LAST + 1;
>         }

>         ...

>         MCO_RET rc;
>         void *mem_buff;

>         mco_db_connect (mydatabase_name, &db);

>         /* if used memory management library is different from mcomconv */
>         #ifdef USE_SHARED_MEMORY
>         mem_buff = 0;
>         #else
>         mem_buff = malloc (TEMP_BUFFER_SIZE);
>         #endif

>         mco_translog_iterate (PIPE_OVEFLOW_FILE, db, iteration_proc, user_ctx, mydatabase_get_dictionary (), mem_buff, TEMP_BUFFER_SIZE);

> ### Endian conversion

> It is possible to mix little-endian and big-endian architectures using Transaction Logging. In other words, applications can create a transaction log on a little-endian system and replay the log to a big-endian system, or vice-versa. There is proper auto-detection and auto-conversion which does not require further configuration or user intervention. However, note that mixing of different word-size platforms, i,e, `x32` vs. `x64`, is not allowed. Also, note that the same transaction manager (MURSIW or MVCC) must be used in both the writer (log creator) and reader applications.

> ## Dual out logging mode

> eXtremeDB Transaction Logging allows writing a copy of the transaction log to a file while the pipe mode is used as a transport for exporting transaction data. This file may be used, for instance, to restore a database or to store it in log form on persistent media. The stored log file may then be processed by the API function `mco_translog_apply()`. Or it can be iterated by functions `mco_translog_iterate()` and `mco_translog_play()`. Dual out logging is enabled by specifying flag `MCO_TRANSLOG_DUALOUT` in addition to flag ``MCO_TRANSLOG_PIPE``. Field `dual_log_path` of structure `mco_TL_start_data` specifies the destination log file. (Sample tlogiterdualout demonstrates this functionality.)

> ## Dynamic Pipes

> This feature (in eXtremeDB Transaction Logging version 6.5 and later) allows applications to create / remove pipes as well as connect to and disconnect from them at runtime. Database events are logged to the pipe and the connected applications are able to process these logs as necessary; they may also stop processing them and remove the associated resources (pipes and connections) at will. The following application scenario is supported through the dynamic pipes:

> > 1. The primary process creates a shared memory database and starts the transaction logging.

> > 2. Secondary processes connect to the database at any given time, create a pipe using this connection and then process the log that the primary application writes into the pipe.

> In reality the secondary processes are asynchronously listening to the database events written into the log, and also have access to the content of the modified database objects that triggered those events. The secondary processes are free to disconnect from the database, and destroy their pipe at any time.

> Naturally if the primary process stops the transaction logging via `mco_translog_stop()`, all connected readers get the appropriate return code. The primary application does not receive or process any notifications from the secondary applications, but merely commits its transactions as usual, logging events if required. Multiple pipes can be created at a time and multiple readers can be connected to those pipes.

> ### Implementation

> > 1. If the flag `MCO_TRANSLOG_DYNAMIC_PIPE` for `mco_translog_start()` is set, then readers can attach and detach to pipes at any time without affecting the writers (threads from which the transactions are committed). Each reader must create its own pipe - two or more readers cannot be connected to the same pipe.

> > 2. The API `mco_translog_play_ex()` is similar to `mco_translog_play()`, except that it takes a single structure `mco_TL_play_params_t` as parameter. The fields of the structure correspond to parameters of `mco_translog_play()` with one additional field:

> >         mco_device_t *pipe_device;

> > The `pipe_device` is a descriptor of the pipe device to be used by the readers. If the value of the `pipe_device` is `NULL` then the transaction logging runtime chooses the first pipe that does not have any connected readers. The same happens when the old style `mco_translog_play()`and `mco_translog_iterate()` APIs are used.

> > 3. The `mco_translog_play_params_init()` function zeroes out the `mco_TL_play_params_t` structure.

> > 4. The runtime option `MCO_RT_MAX_DYNAMIC_PIPES` has the default value of `0`. This value defines the maximum number of the `PIPE_BUF` type devices that can be added through `mco_db_extend_dev()`. This limit does not include the pipe devices created via `mco_db_open_dev()`. In other words, if the application sets `MCO_RT_MAX_DYNAMIC_PIPES` to 3 and defines 5 pipe devices at the time the database was created, then it will be able to have up to 8 pipe devices at any moment in time.

> ## Using event handlers during iteration of log file or pipe

> If the database schema defines events, their handlers may be registered and fired during the reading of the log file or pipe in addition to or instead of iterating the objects themselves. This technique is called 'persistent events queue' and can be used as a method of asynchronous event handling by the current process or by a separate process.

> To use this feature, the database schema must define one or more events, and the application must define the handlers for these events, and a function which registers the handlers must be defined in the application. Then function `mco_translog_play()` is used instead of function `mco_translog_iterate()` to iterate the log. (Sample tlogiterevents demonstrates this functionality.)

> ### Filtering data stored in the log by events

> An events mask may be specified to limit the data written to the logging stream transferred to the log file or pipe. Note that this mode does not allow the use of the log for database recovery due to the filtering. However, it can be useful for specific purposes to reduce the amount of data transferred to the log. For example, if event handlers are used on the reader's side and it is not necessary to transfer all of the data to the log except notification of the event itself.

> Flag ``MCO_TRANSLOG_EVENT_MASK`` is specified to enable this feature and the `event_mask` field of structure `mco_TL_start_data` is used to define the desired mask. There are two predefined mask values: `MCO_LOG_MASK_ALL_CHANGES` stores all data in the log, which is equivalent to disabling events mask filtering; `MCO_LOG_MASK_ALL_EVENTS` stores only the data related to events defined in the database schema. If these predefined values are not used, the mask must be composed from the event handler identifiers generated by the database schema compiler. (Note that event handler identifiers are not zero-based, i.e. they are natural numbers starting from one, and must be decremented by one to use it in a mask. For example `new`, `delete` and `delete all` events could be specified in a mask as follows:

>         tl_parms.event_mask = (1 << (MCO_EVENT_newEvent - 1)) |
>                         (1 << (MCO_EVENT_deleteEvent - 1)) |
>                         (1 << (MCO_EVENT_deleteAllEvent - 1));

> (Sample tlogiterevents demonstrates the use of filtering.)

> ## User-defined iteration callback

> Consider the following schema:

>         class Record_A
>         {
>             uint4 key;
>         };

>         class Record_B
>         {
>             uint4 key;
>         };

> Part of the code generated by the schema compiler would be:

>         typedef struct Record_A_   { MCO_Hf h; }  Record_A;
>         #define                  Record_A_code          1

>         typedef struct Record_B_   { MCO_Hf h; }  Record_B;
>         #define                  Record_B_code          2

> The user-defined callback `iteration_proc` is called by `mco_translog_iterate()` for each object one by one. This `iteration_proc` has the following definition:

> Prototype | `MCO_RET` | `iteration_proc (` | `mco_trans_h trans, MCO_Hf* obj, int cid, int obj_state, void* user_ctx )`
> | Arguments | `trans` | Transaction handle. (It is possible to get the transaction number by calling `mco_trans_no()` with this handle.) Could be 0 to indicate that function `mco_db_clean()` was called in source database. See sample tlogiterpipe for details.
> |  | `obj` | Object handle of the object that was actually created, deleted or updated inside the transaction. This handle may be used to access the fields of the object with calls like: `Record_A_key_get((Record_A*)obj, &key);``Record_B_key_get((Record_B*)obj, &key);` Could be 0 if end of transaction state is indicated. (See section `Option Mark_Last_Object` for details).
> |  | `cid` | The class Id of this object. This will have the same value as the definitions `Record_A_code` or `Record_B_code`. Compare this code to `Record_A_code` or `Record_B_code` to determine which type of object is present before attempting to access its fields.
> |  | `obj_state` | The type of operation that was performed on the object: `MCO_TRANS_OBJ_ALL_DELETED` – delete all objects of this class. `MCO_TRANS_OBJ_DELETED` – the object was deleted. `MCO_TRANS_OBJ_CREATED` – the object was created. 0 - the object was updated.
> |  | `user_ctx` | A pointer to user data passed into the function `mco_trans_iterate()`.

> Note: If the `obj_state` value is `MCO_TRANS_OBJ_ALL_DELETED`, then obj is the first object of its class in the database for which the function `classname_delete_all()` was called inside a transaction. For example: `Record_A_delete_all()`. If there were no such objects in the database at the moment `classname_delete_all()` was called, then the current iteration callback will not be called. So it is impossible to have an empty `obj` argument.

> #### Example of user-defined callback implementation for the above schema:

>         MCO_RET iteration_proc (mco_trans_h trans, MCO_Hf* obj, int cid, int obj_state, void* user_ctx)
>         {
>             mco_trans_counter_t trans_no;
>             if (trans == 0)
>             {
>                 /* handle call of mco_db_clean called in source DB */
>                 return MCO_S_OK;
>             }

>             mco_trans_no (trans, &trans_no);
>             if (cid == Record_A_code)
>             {
>                 uint4 key;

>                 if (obj_state == MCO_TRANS_OBJ_ALL_DELETED)
>                 {
>                     /* Delete all objects of class Record_A */
>                 }
>                 else if (obj_state == MCO_TRANS_OBJ_DELETED)
>                 {
>                     /* Delete current object by the key */
>                     Record_A_key_get ((Record_A *)obj, &key);
>                 }
>                 else if (obj_state == MCO_TRANS_OBJ_CREATED)
>                 {
>                     /* Create new object with the key */
>                     Record_A_key_get ((Record_A *)obj, &key);
>                 }
>                 else
>                 {
>                     /* Update object with the key */
>                     Record_A_key_get ((Record_A *)obj, &key);
>                 }
>             }
>             else if (cid == Record_B_code)
>             {
>                 uint4 key;

>                 if (obj_state == MCO_TRANS_OBJ_ALL_DELETED)
>                 {
>                     /* Delete all objects of class Record_B */
>                 }
>                 else if (obj_state == MCO_TRANS_OBJ_DELETED)
>                 {
>                     /* Delete current object by the key */
>                     Record_B_key_get ((Record_B *)obj, &key);
>                 }
>                 else if (obj_state == MCO_TRANS_OBJ_CREATED)
>                 {
>                     /* Create new object with the key */
>                     Record_B_key_get ((Record_B *)obj, &key);
>                 }
>                 else
>                 {
>                     /* Update object with the key */
>                     Record_B_key_get ((Record_B *)obj, &key);
>                 }
>             }

>             return MCO_S_OK;
>         }

> ## User-defined register event handlers callback

> Consider the following schema:

>         class Record
>         {
>             uint4 key;
>             event <new>         newEvent;
>         };

> Part of the code generated by the schema compiler would be:

>         #define MCO_EVENT_newEvent  1
>         typedef MCO_RET (*mco_newEvent_handler)( mco_trans_h t, Record * obj, MCO_EVENT_TYPE et, /*INOUT*/ void *param);
>         MCO_RET  mco_register_newEvent_handler       ( mco_trans_h t, mco_newEvent_handler handler, void * param );
>         MCO_RET  mco_unregister_newEvent_handler     ( mco_trans_h t, mco_newEvent_handler handler);

> The user-defined callback `register_callback()` is called by `mco_translog_play()` just before starting the iteration of objects. This `register_callback()` has the following definition:

> Prototype | `MCO_RET` | `register_callback (` | `mco_trans_h trans, void* user_ctx )`
> | Arguments | `trans` | Transaction handle. This handle should be passed to event registering function.
> |  | `user_ctx` | A pointer to user data passed into argument `regevent_user_ctx` of the function `mco_trans_play()`.

> #### Example of user-defined register callback implementation for the above schema:

>         MCO_RET register_callback(mco_trans_h t, void *param)
>         {
>             return mco_register_newEvent_handler(t, my_newEvent_handler, param);
>         }

>         MCO_RET my_newEvent_handler( mco_trans_h t, Record * obj, MCO_EVENT_TYPE et, /*INOUT*/ void *param)
>         {
>             ev_stat_t *s = (ev_stat_t *)param;
>             printf("my_newEvent_handler %d\n", ++(s->new_cnt));
>             return MCO_S_OK;
>         }

> ## Precautions for a potential system failure

> If Transaction Logging is employed in a multi-process scenario using pipes, and one of the processes crashes (for example the reader) then the writer process may be suspended waiting on internal locks used in a pipe. To resolve this situation a “sniffer” task can be run alongside the Transaction Logging processes; the `sniffer` task will clean up locks left over from the `dead` connections left by the crashed processes. (Please refer to the “Database Recovery from Failed Processes” section of the eXtremeDB User Guide for details regarding the `sniffer` API.)

> Regardless of the method of Transaction Logging (TL) being used (`pipe`, `multi-pipe` or `dynamic-pipe`), the following steps are recommended:

> > 1. Run sniffer in the context of a separate thread in the main application:

> >         #define SNIFFER_INTERVAL 100
> >         MCO_RET sniffer_callback(mco_db_h db, void* context, mco_trans_counter_t trans_no)
> >         {
> >             SAMPLE_OS_TASK_ID pid = *(SAMPLE_OS_TASK_ID *)context;
> >             if ( sample_os_task_id_check( pid ) == 0 )
> >             {
> >                 return MCO_S_OK;
> >             }
> >             printf("Process %d is crashed\n", pid);
> >             return MCO_S_DEAD_CONNECTION;
> >         }
> >         void sniffer_loop( sample_task_t * descriptor )
> >         {
> >             mco_db_h db;
> >             SAMPLE_OS_TASK_ID pid = sample_os_task_id_get();

> >             /* Connect using mco_db_connect_ctx() and pass &pid as parameter */
> >             MCO_RET rc = mco_db_connect_ctx(db_name, &pid, &db);
> >             if ( MCO_S_OK == rc )
> >             {

> >                 /* Descriptor->stopped flag is set to 1 by sample_stop_task() in main thread*/
> >                 while ( MCO_S_OK == rc && descriptor->stopped == 0 )
> >                 {
> >                     rc = mco_db_sniffer(db, sniffer_callback, MCO_SNIFFER_INSPECT_ACTIVE_CONNECTIONS);
> >                     sample_sleep(SNIFFER_INTERVAL);
> >                 }
> >                 mco_db_disconnect(db);
> >             }
> >         }

> > Note that the sniffer is executed in the task that contains the `sniffer_loop()` function. The callback itself merely makes sure that all tasks connected to the database are alive (see for example samples/native/core/19-recovery/sniffer).

> > 2a) For the `pipe` case the main program will initialize the TL processing and watch for any errors reported by TL. If any errors are reported, the main application will terminate the TL processing (which will free all locks set by the TL runtime) and, if necessary, restart TL to allow other TL clients to continue processing the pipe, for example:

> >         ...

> >         /* allocate pipe memory device */
> >         dev[1].type       = MCO_MEMORY_NAMED;
> >         dev[1].assignment = MCO_MEMORY_ASSIGN_PIPE_BUF;
> >         sprintf( dev[1].dev.named.name, "%s-pipe", db_name );
> >         dev[1].size       = DATABASE_SIZE / 2;
> >         dev[1].dev.named.flags = 0;
> >         dev[1].dev.named.hint  = 0;
> >         ...

> >         /* Set up and run transaction logging (pipe is created here) */
> >         tl_parms.flags = MCO_TRANSLOG_ITERABLE | MCO_TRANSLOG_PIPE;
> >         tl_parms.disk_page_size = PSTORAGE_PAGE_SIZE;
> >         rc = mco_translog_start( connection, 0, &tl_parms );

> >         /* Wait for TL-client connection */
> >         printf("\n\n\tWaiting for log reader ...\n" );
> >         while(1)
> >         {
> >             mco_TL_current_info_t tl_info;
> >             mco_translog_get_info ( connection, &tl_info );
> >             if (tl_info.pipe_readers_connected)
> >                 break;
> >             else
> >                 sample_sleep (100);
> >         }
> >         ...
> >         while ( data processing )
> >         {
> >             while (1)
> >             {
> >                 /* process database transaction */
> >                 if (rc == MCO_S_OK)
> >                 {
> >                     ...
> >                 }
> >                  else if (rc == MCO_E_TL_IO_ERROR || rc == MCO_E_TL_PIPE_TERM )
> >                 {
> >                     /* TL processing error detected */

> >                     /* force logging to stop */
> >                     rc = mco_translog_terminate( connection );

> >                     /* wait for TL connection */
> >                     printf("\n\n\tWaiting for log reader ...\n" );
> >                     while(1)
> >                     {
> >                         mco_TL_current_info_t tl_info;
> >                         mco_translog_get_info ( connection, &tl_info );
> >                         if (tl_info.pipe_readers_connected)
> >                             break;
> >                         else
> >                             sample_sleep (100);
> >                     }

> >                     /* restart TL processing  */
> >                     tl_parms.flags |= MCO_TRANSLOG_RESTART;
> >                     rc = mco_translog_start( connection, 0, &tl_parms );
> >                     sample_rc_check( "\tRe-start logging", rc );
> >                     break;
> >                 }
> >                 else
> >                     ...
> >             }
> >         }

> > 2b. For the `Multi-pipe` case the TL handling is similar, except that it is necessary to create several pipe-devices:

> >         ...
> >         for ( i=1; i<=PIPES_COUNT; i++ ) {
> >             dev[i].type       = MCO_MEMORY_NAMED;
> >             dev[i].assignment = MCO_MEMORY_ASSIGN_PIPE_BUF;
> >             sprintf( dev[i].dev.named.name, "%s-pipe-%u", db_name, i );
> >             dev[i].size       = DATABASE_SIZE / 2;
> >             dev[i].dev.named.flags = 0;
> >             dev[i].dev.named.hint  = 0;
> >         }
> >         ...
> >         rc = mco_translog_start( connection, 0, &tl_parms );

> >         sample_rc_check( "\tStart logging", rc );

> >         /* Wait while iterator thread is alive */
> >         printf("\n\n\tWaiting for log reader ...\n" );
> >         while(1)
> >         {
> >             mco_TL_current_info_t tl_info;
> >             mco_translog_get_info ( connection, &tl_info );
> >             if (tl_info.pipe_readers_connected == PIPES_COUNT)
> >                 break;
> >             else
> >                 sample_sleep (100);
> >         }
> >         ...
> >         while ( data processing )
> >         {
> >             while (1)
> >             {
> >                 /* process database transaction */
> >                 if (rc == MCO_S_OK)
> >                 {
> >                     ...
> >                 }
> >                 else if (rc == MCO_E_TL_IO_ERROR || rc == MCO_E_TL_PIPE_TERM )
> >                 {
> >                     /* TL processing error detected */
> >                     /* force logging to stop */
> >                     rc = mco_translog_terminate( connection );

> >                     /* wait for TL connection */
> >                     printf("\n\n\tWaiting for log reader ...\n" );
> >                     while(1)
> >                     {
> >                         mco_TL_current_info_t tl_info;
> >                         mco_translog_get_info ( connection, &tl_info );
> >                         if (tl_info.pipe_readers_connected == PIPES_COUNT)
> >                             break;
> >                         else
> >                             sample_sleep (100);
> >                     }
> >                     /* restart TL processing  */
> >                     tl_parms.flags |= MCO_TRANSLOG_RESTART;
> >                     rc = mco_translog_start( connection, 0, &tl_parms );
> >                     sample_rc_check( "\tRe-start logging", rc );
> >                     break;
> >                 } else
> >                 ...
> >             }
> >         }

> > 2c. The `dynamic-pipes` case is trivial, for example:

> >         ...
> >         /* Set maximum number of dynamic pipe segments */
> >         mco_runtime_setoption(MCO_RT_MAX_DYNAMIC_PIPES, PIPES_COUNT);
> >         ...
> >         /* Set default database parameters */
> >         mco_db_params_init ( &db_params );

> >         /* Customize the params according to the application */
> >         db_params.mem_page_size            = MEMORY_PAGE_SIZE;
> >         db_params.disk_page_size           = 0;            /* Pure in-memory database */
> >         db_params.db_max_connections       = 10 + PIPES_COUNT;
> >         db_params.connection_context_size  = sizeof(SAMPLE_OS_TASK_ID);
> >         ...

> >         /* Set up and run transaction logging (pipe is created here) */
> >         tl_parms.flags = MCO_TRANSLOG_ITERABLE | MCO_TRANSLOG_PIPE |
> >         MCO_TRANSLOG_DYNAMIC_PIPE | MCO_TRANSLOG_SYNC_INSTANTLY;
> >         tl_parms.disk_page_size = PSTORAGE_PAGE_SIZE;
> >         rc = mco_translog_start( connection, 0, &tl_parms );
> >         sample_rc_check( "\tStart logging", rc );

> >         /* Wait while iterator thread is alive */
> >         printf("\n\n\tWaiting for log reader ...\n" );
> >         while(1)
> >         {
> >             mco_TL_current_info_t tl_info;
> >             mco_translog_get_info ( connection, &tl_info );
> >             if (tl_info.pipe_readers_connected)
> >                 break;
> >             else
> >                 sample_sleep (100);
> >         }
> >         ...

> > ## Possible Data Relay Deadlock

> > There are some possible situations where data relay deadlock may happen. For instance:

> > - When there are two database (or more) connections, the first connection might flood the database with new records while the second connection is used to start the transaction logging reader (e.g. calling `mco_translog_play()` etc.). Then inside a user-defined reader callback function there might be some data processing that results in writing back into the same database from this second connection. Under heavy load it is possible that a write transaction in a user-defined reader callback (on the second connection) could not be completed and is locked because there is no room in the pipe to write data. But at the same time the pipe could not be read (to free some space in it) because the reading procedure is performing in the same thread where commit operation is already blocked.
> > - Under heavy load without a pipe overflow file when the thread committing objects in response to reading by the iterator fails to perform a commit because the pipe is full. It blocks waiting for the reader to free space in the pipe. But the reader thread fails to do that being blocked itself waiting for the first thread to write and free the queue which is also limited and full. Actually any of the threads (connections) that is trying to commit a transaction will be blocked if the pipe is full and the reader is not able to read data as reader thread is blocked in one of the hung commits itself.

> > This is non-obvious yet natural behavior of the closed loop system (commit - log - pipe - reader's analyzer - commit again). Trying to split the pipe reading procedure and the commit of new data which is some derivative of read data (eg. commit - log - pipe - reader's analyzer - some data queue - another application thread and DB connection - commit again) may cause deadlock. Basically if the application is committing some data to a database which is the result of reading the transaction logging pipe, dead-lock is possible under heavy load situations.

> > This can be moderated at the application level by artificially throttling the initial commits. In fact, initially the Data Relay feature was designed to transfer data from eXtremeDB to another DBMS or to some data processor. So it was not assumed that processed data would be committed back to eXtremeDB. Fortunately the pipe overflow files resolve this problem of "data loop-back". The important fact is the only reliable way to avoid deadlock is to use a pipe overflow file (which does imply a penalty to the overall performance).

> > ## Possible Error Conditions for MCO_E_TL_PIPE_TERM

> > There are two cases when error code `MCO_E_TL_PIPE_TERM` could be returned both by a reader call to `mco_translog_iterate()` and a writer performing a transaction commit operation. The two scenarios could be as follows:

> > ### Case A: MCO_E_TL_PIPE_TERM in reader and writer

> > > 1. Call `mco_translog_start()` to start logging

> > > 2. Call `mco_translog_iterate()` in an iteration thread to start reading

> > > 3. The user-defined iteration procedure returns an error code (anything but `MCO_S_OK`) to terminate function `mco_translog_iterate()`

> > > 4. When function `mco_translog_iterate()` returns with the code specified by iteration procedure, the pipe gets terminated

> > > 5. The writer tries to commit a transaction and function `mco_trans_commit()` will return `MCO_E_TL_PIPE_TERM` because the pipe has already been terminated

> > > 6. Start `mco_translog_iterate()` again without first calling function `mco_translog_stop()`

> > > 7. Now `mco_translog_iterate()` returns `MCO_E_TL_PIPE_TERM` because the pipe has terminated

> > ### Case B: `MCO_E_TL_PIPE_TERM` in reader and writer after `mco_translog_terminate()`

> > > 1 Call `mco_translog_start()` to start logging

> > > 2 Call `mco_translog_iterate()` in an iteration thread to start reading

> > > 3 Call `mco_translog_terminate()` in a writer thread (or the main instance) and the pipe gets terminated

> > > 4 Now function `mco_translog_iterate()` returns `MCO_E_TL_PIPE_TERM`

> > > 5 Trying to commit a transaction causes function `mco_trans_commit()` to return `MCO_E_TL_PIPE_TERM` because the pipe has already terminated

> > Note that if the user-defined iteration callback always returns `MCO_S_OK`, then scenario A is not possible; scenario B is possible if function `mco_translog_terminate()` is called when error code `MCO_E_TL_IO_ERROR` is returned by `mco_trans_commit()`.