Write a loader with a replica preload controller

Program guide > Programming with system APIs and plug-ins > Persistent store plug-ins

Write a loader with a replica preload controller

A Loader with a replica preload controller is a Loader that implements the ReplicaPreloadController interface in addition to the Loader interface.

The ReplicaPreloadController interface is designed to provide a way for a replica that becomes the primary shard to know whether the previous primary shard has completed the preload process. If the preload is partially completed, the information to pick up where the previous primary left is provided. With the ReplicaPreloadController interface implementation, a replica that becomes the primary continues the preload process where the previous primary left and continues to finish the overall preload.

In a distributed WebSphere eXtreme Scale environment, a map can have replicas and might preload large volume of data during initialization. The preload is a Loader activity and only occurs in the primary map during initialization. The preload might take a long time to complete if a large volume of data is preloaded. If the primary map has completed large portion of preload data, but is stopped for unknown reason during initialization, a replica becomes the primary. In this situation, the preloaded data that was completed by the previous primary is lost because the new primary normally performs an unconditional preload. With an unconditional preload, the new primary starts the preload process from the beginning and the previous preloaded data is ignored. If you want the new primary to pick up where the previous primary left during preload process, provide a Loader that implements the ReplicaPreloadController interface. For more information see the API documentation.

For information about Loaders, see Loaders. If you are interested in writing a regular Loader plug-in, see Write a loader.

The ReplicaPreloadController interface has the following definition:

public interface ReplicaPreloadController
{
    public static final class Status
    {
        static public final Status PRELOADED_ALREADY =
                            new Status(K_PRELOADED_ALREADY);
        static public final Status FULL_PRELOAD_NEEDED =
                            new Status(K_FULL_PRELOAD_NEEDED);
        static public final Status PARTIAL_PRELOAD_NEEDED =
                            new Status(K_PARTIAL_PRELOAD_NEEDED);
    }

    Status checkPreloadStatus(Session session,
                BackingMap bmap);
}

The following sections discuss some of the methods of the Loader and ReplicaPreloadController interface.

checkPreloadStatus method

When a Loader implements ReplicaPreloadController interface, the checkPreloadStatus method is called before the preloadMap method during map initialization. The return status of this method determines if the preloadMap method is called. If this method returns Status#PRELOADED_ALREADY, the preload method is not called. Otherwise, the preload method runs. Because of this behavior, this method should serve as the Loader initialization method. You must initialize the Loader properties in this method. This method must return the correct status, or the preload might not work as expected.

public Status checkPreloadStatus(Session session,
                            BackingMap backingMap) {
        // When a loader implements ReplicaPreloadController interface,
             //    this method will be called before preloadMap method during
             // map initialization. Whether the preloadMap method will be
             //    called depends on teh returned status of this method. So, this
             // method also serve as Loader's initialization method. This method
             // has to return the right staus, otherwise the preload may not
             // work as expected.

        // Note: must initialize this loader instance here.
        ivOptimisticCallback = backingMap.getOptimisticCallback();
        ivBackingMapName = backingMap.getName();
        ivPartitionId = backingMap.getPartitionId();
        ivPartitionManager = backingMap.getPartitionManager();
        ivTransformer = backingMap.getObjectTransformer();
        preloadStatusKey = ivBackingMapName + "_" + ivPartitionId;

        try {
            // get the preloadStatusMap to retrieve preload status that
                     // could be set by other JVMs.
            ObjectMap preloadStatusMap = session.getMap(ivPreloadStatusMapName);

            // retrieve last recorded preload data chunk index.
            Integer lastPreloadedDataChunk = (Integer) preloadStatusMap
                                    .get(preloadStatusKey);

            if (lastPreloadedDataChunk == null) {
                preloadStatus = Status.FULL_PRELOAD_NEEDED;
            } else {
                preloadedLastDataChunkIndex = lastPreloadedDataChunk.intValue();
                if (preloadedLastDataChunkIndex == preloadCompleteMark) {
                    preloadStatus = Status.PRELOADED_ALREADY;
                } else {
                    preloadStatus = Status.PARTIAL_PRELOAD_NEEDED;
                }
            }

            System.out.println("TupleHeapCacheWithReplicaPreloadControllerLoader.
                                checkPreloadStatus() 
                    -> map = " + ivBackingMapName + ", preloadStatusKey = " + preloadStatusKey
                    + ", retrieved lastPreloadedDataChunk =" + lastPreloadedDataChunk + ", 
                                    determined preloadStatus = "
                    + getStatusString(preloadStatus));

        } catch (Throwable t) {
            t.printStackTrace();
        }

        return preloadStatus;
    }

preloadMap method

Run the preloadMap method depends on the returned result from checkPreloadStatus method. If the preloadMap method is called, it typically must retrieve preload status information from designated preload status map and determine how to proceed. Ideally, the preloadMap method should know if the preload is partially complete and exactly where to start. During the data preload, the preloadMap method should update the preload status on the designated preload status map. The preload status that is stored in the preload status map is retrieved by the checkPreloadStatus method when it needs to check the preload status.

public void preloadMap(Session session, BackingMap backingMap)
                    throws LoaderException {
        EntityMetadata emd = backingMap.getEntityMetadata();
        if (emd != null && tupleHeapPreloadData != null) {
            // The getPreLoadData method is similar to fetching data
                     // from database. These data will be push into cache as
                     // preload process.
            ivPreloadData = tupleHeapPreloadData.getPreLoadData(emd);

            ivOptimisticCallback = backingMap.getOptimisticCallback();
            ivBackingMapName = backingMap.getName();
            ivPartitionId = backingMap.getPartitionId();
            ivPartitionManager = backingMap.getPartitionManager();
            ivTransformer = backingMap.getObjectTransformer();
            Map preloadMap;

            if (ivPreloadData != null) {
                try {
                    ObjectMap map = session.getMap(ivBackingMapName);

                    // get the preloadStatusMap to record preload status.
                    ObjectMap preloadStatusMap = session.
                                            getMap(ivPreloadStatusMapName);

                    // Note: when this preloadMap method is invoked, the
                                    // checkPreloadStatus has been called, Both preloadStatus
                                    // and preloadedLastDataChunkIndex have been set. And the
                                    // preloadStatus must be either PARTIAL_PRELOAD_NEEDED 
                                    // or FULL_PRELOAD_NEEDED that will require a preload again.

                    // If large amount of data will be preloaded, the data usually
                                    // is divided into few chunks and the preload process will
                                    // process each chunk within its own tran. This sample only
                                    // preload few entries and assuming each entry represent a chunk.
                    // so that the preload process an entry in a tran to simulate
                                    // chunk preloading.

                    Set entrySet = ivPreloadData.entrySet();
                    preloadMap = new HashMap();
                    ivMap = preloadMap;

                    // The dataChunkIndex represent the data chunk that is in
                                    // processing
                    int dataChunkIndex = -1;
                    boolean shouldRecordPreloadStatus = false;
                    int numberOfDataChunk = entrySet.size();
                    System.out.println("    numberOfDataChunk to be preloaded = " 
                                            + numberOfDataChunk);

                    Iterator it = entrySet.iterator();
                    int whileCounter = 0;
                    while (it.hasNext()) {
                        whileCounter++;
                        System.out.println("preloadStatusKey = " + preloadStatusKey
                                                        + " , 
                                                whileCounter = " + whileCounter);

                        dataChunkIndex++;

                        // if the current dataChunkIndex
<= preloadedLastDataChunkIndex
                        // no need to process, becasue it has been preloaded by
                                            // other JVM before. only need to process dataChunkIndex
                                            // > preloadedLastDataChunkIndex
                        if (dataChunkIndex
<= preloadedLastDataChunkIndex) {
                            System.out.println("ignore current dataChunkIndex =
                                                            " + dataChunkIndex + " that has been previously
                                                            preloaded.");
                            continue;
                        }

                        // Note: This sample simulate data chunk as an entry.
                        // each key represent a data chunk for simplicity.
                        // If the primary server or shard stopped for unknown
                                          // reason, the preload status that indicates the progress
                                          // of preload should be available in preloadStatusMap. A
                                            // replica that become a primary can get the preload status
                                            // and determine how to preload again.
                        // Note: recording preload status should be in the same
                                          // tran as putting data into cache; so that if tran
                                            // rollback or error, the recorded preload status is the
                                            // actual status.

                        Map.Entry entry = (Entry) it.next();
                        Object key = entry.getKey();
                        Object value = entry.getValue();
                        boolean tranActive = false;

                        System.out.println("processing data chunk. map = " +
                                                    this.ivBackingMapName + ", current dataChunkIndex = " +
                                                    dataChunkIndex + ", key = " + key);

                        try {
                            shouldRecordPreloadStatus = false; // re-set to false
                            session.beginNoWriteThrough();
                            tranActive = true;

                            if (ivPartitionManager.getNumOfPartitions() == 1) {
                                // if just only 1 partition, no need to deal with
                                                         // partition.
                                // just push data into cache
                                map.put(key, value);
                                preloadMap.put(key, value);
                                shouldRecordPreloadStatus = true;
                            } else if (ivPartitionManager.getPartition(key) ==
                                                         ivPartitionId) {
                                // if map is partitioned, need to consider the
                                                         // partition key only preload data that belongs
                                                         // to this partition.
                                map.put(key, value);
                                preloadMap.put(key, value);
                                shouldRecordPreloadStatus = true;
                            } else {
                                // ignore this entry, because it does not belong to
                                                         // this partition.
                            }

                            if (shouldRecordPreloadStatus) {
                                System.out.println("record preload status. map = " +
                                                                        this.ivBackingMapName + ", preloadStatusKey = " +
                                                                        preloadStatusKey + ", current dataChunkIndex ="
                                                                        + dataChunkIndex);
                                if (dataChunkIndex == numberOfDataChunk) {
                                    System.out.println("record preload status. map = " +
                                                                              this.ivBackingMapName + ", preloadStatusKey = " +
                                                                                preloadStatusKey + ", mark complete =" +
                                                                                preloadCompleteMark);
                                    // means we are at the lastest data chunk, if commit
                                                                // successfully, record preload complete.
                                                                // at this point, the preload is considered to be done
                                    // use -99 as special mark for preload complete status.

                                    preloadStatusMap.get(preloadStatusKey);

                                    // a put follow a get will become update if the get
                                                                 // return an object, otherwise, it will be insert.
                                    preloadStatusMap.put(preloadStatusKey, new
                                                                            Integer(preloadCompleteMark));

                                } else {
                                    // record preloaded current dataChunkIndex into
                                                                 // preloadStatusMap a put follow a get will become
                                                                 // update if teh get return an object, otherwise, it
                                                                 // will be insert.
                                    preloadStatusMap.get(preloadStatusKey);
                                    preloadStatusMap.put(preloadStatusKey, new
                                                                        Integer(dataChunkIndex));
                                }
                            }

                            session.commit();
                            tranActive = false;

                            // to simulate preloading large amount of data
                            // put this thread into sleep for 30 secs.
                            // The real app should NOT put this thread to sleep
                            Thread.sleep(10000);

                        } catch (Throwable e) {
                            e.printStackTrace();
                            throw new LoaderException("preload failed with
                                                  exception: " + e, e);
                        } finally {
                            if (tranActive && session != null) {
                                try {
                                    session.rollback();
                                } catch (Throwable e1) {
                                    // preload ignoring exception from rollback
                                }
                            }
                        }
                    }

                    // at this point, the preload is considered to be done for sure
                    // use -99 as special mark for preload complete status.
                    // this is a insurance to make sure the complete mark is set.
                    // besides, when partitioning, each partition does not know when
                                    // is its last data chunk. so the following block serves as the
                                    // overall preload status complete reporting.
                    System.out.println("Overall preload status complete -> record
                                                  preload status. map = " + this.ivBackingMapName + ",
                                                  preloadStatusKey = " + preloadStatusKey + ", mark complete =" +
                                                    preloadCompleteMark);
                    session.begin();
                    preloadStatusMap.get(preloadStatusKey);
                    // a put follow a get will become update if teh get return an object,
                    // otherwise, it will be insert.
                    preloadStatusMap.put(preloadStatusKey, new Integer(preloadCompleteMark));
                    session.commit();

                    ivMap = preloadMap;
                } catch (Throwable e) {
                    e.printStackTrace();
                    throw new LoaderException("preload failed with exception: " + e, e);
                }
            }
        }
    }

Preload status map

Use a preload status map to support the ReplicaPreloadController interface implementation. The preloadMap method should always check the preload status stored in the preload status map first and update the preload status in the preload status map whenever it pushes data into the cache. The checkPreloadStatus method can retrieve the preload status from preload status map, determine the preload status, and return the status to its caller. The preload status map should be in the same mapSet as other maps that have replica preload controller Loaders.