kmrckpt.h

Go to the documentation of this file.

/* kmrckpt.h (2014-04-01) */
/* Copyright (C) 2012-2016 RIKEN AICS */

#ifndef _KMRCKPT_H
#define _KMRCKPT_H

/** \file kmrckpt.h Checkpoint/Restart Support.
    Checkpoint/Restart provides two modes; SAVE_ALL, SAVE_SELECTIVE.
    In SAVE_ALL mode, checkpoints of all key-value stores are taken.
    IN SAVE_SELECTIVE mode, only checkpoints of user-specified key-value
    stores are taken. */

#define KMR_CKPT_DIRNAME       "ckptdir"
#define KMR_CKPT_FNAME_PREFIX  "kmrckpt"
#define KMR_CKPT_PATHLEN       512
#define KMR_CKPT_DIRLEN        256
#define KMR_CKPT_MSGLEN        512


/** Checkpoint modes.  KMR_CKPT_ALL takes checkpoints of all key-value
    stores.  KMR_CKPT_SELECTIVE takes checkpoints of key-value stores
    that user specified. */
enum kmr_ckpt_mode {
    KMR_CKPT_ALL,
    KMR_CKPT_SELECTIVE
};

/* An item in a list */
struct kmr_ckpt_list_item {
    void *val;
    struct kmr_ckpt_list_item *next;
    struct kmr_ckpt_list_item *prev;
};

/* Maximum size of a list */
#define KMR_CKPT_LIST_MAX 1000

/* allocator for items in a list */
typedef void * (*kmr_ckpt_list_alocfn_t)(void *);
/* deallocator for items in a list */
typedef void (*kmr_ckpt_list_freefn_t)(void *);
/* comparater for items in a list */
typedef int (*kmr_ckpt_list_compfn_t)(void *, void *);

/* Linked list */
struct kmr_ckpt_list {
    struct kmr_ckpt_list_item *head;
    struct kmr_ckpt_list_item *tail;
    kmr_ckpt_list_alocfn_t alocfn;
    kmr_ckpt_list_freefn_t freefn;
    kmr_ckpt_list_compfn_t compfn;
    long size;                      /* size of stored items */
};

/* A collection that stores kvs transitions.  The items in this collection
   is struct kmr_ckpt_operation. */
struct kmr_ckpt_kvs_chains {
    struct kmr_ckpt_list *chainlst;
    int chainlst_size;
};

/* Record of an operation to input and output KVS */
struct kmr_ckpt_operation {
    long op_seqno;
    long kvi_id;
    long kvo_id;
};

/* Record of map/reduce operation start kv position */
struct kv_position {
    long op_seqno;
    long start_from;
};

/* Checkpoint context data */
struct kmr_ckpt_ctx {
    char ckpt_dname[KMR_CKPT_DIRLEN];  /* checkpoint directory name */
    int prev_mode;                     /* checkpoint mode of previous run */
    FILE *ckpt_log_fp;                 /* ckpt log file pointer */
    long progress_counter;             /* MapReduce progress counter */
    long prev_progress;                /* progress of previous run */
    long prev_global_progress;         /* global progress of previous run */
    long cur_kvi_id;                   /* ID of current targeted KVI */
    long cur_kvo_id;                   /* ID of current targeted KVO */
    FILE *ckpt_data_fp;                /* file pointer to a ckpt data file
                                          that is currently targeted */
    long saved_element_count;          /* count of key-values in a KVS
                                          that are already written to a
                                          ckpt data file */
    void *saved_adding_point;          /* key-value position in a KVS from
                                          which writing to a ckpt data file
                                          starts */
    void *saved_current_block;         /* key-values block in a KVS from
                                          which writing to a ckpt data file
                                          starts */
    struct kv_position *kv_positions;  /* start kv position in map/reduce */
    int kv_positions_count;            /* count of kv_positions */
    int lock_id;                       /* ckpt lock id */
    int lock_counter;                  /* ckpt lock counter */
    _Bool initialized;                 /* 1 if ckpt_ctx is initialised */
    _Bool slct_cur_take_ckpt;          /* [selective mode]
                                          if 1, ckpt data of the output KVS
                                          generated by current operation will
                                          be taken */
    struct kmr_ckpt_list *slct_skip_ops;
                                       /* [selective mode]
                                          list of id of operation that can
                                          be skepped when restarted */
};

/* Checkpoint data file header */
struct kmr_ckpt_data {
    char   magic[8];            /* magic number */
    char   info[4];             /* misc info (ex. compressed etc.) */
    int    nprocs;              /* MPI communicator size*/
    int    rank;                /* MPI rank */
    long   kvs_id;              /* ID of saved kvs */
    enum kmr_kv_field key_data;
    enum kmr_kv_field value_data;
    struct kmr_kvs_entry data[1];
};

/* Log entry type */
enum kmr_ckpt_log_state {
    KMR_CKPT_LOG_WHOLE_START,
    KMR_CKPT_LOG_WHOLE_FINISH,
    KMR_CKPT_LOG_BLOCK_START,
    KMR_CKPT_LOG_BLOCK_ADD,
    KMR_CKPT_LOG_BLOCK_FINISH,
    KMR_CKPT_LOG_INDEX_START,
    KMR_CKPT_LOG_INDEX_ADD,
    KMR_CKPT_LOG_INDEX_FINISH,
    KMR_CKPT_LOG_DELETE_START,
    KMR_CKPT_LOG_DELETE_FINISH,
    KMR_CKPT_LOG_DELETABLE,
    KMR_CKPT_LOG_PROGRESS,
    KMR_CKPT_LOG_SKIPPED,
    KMR_CKPT_LOG_LOCK_START,
    KMR_CKPT_LOG_LOCK_FINISH
};

/* Checkpoint log entry */
struct kmr_ckpt_log_entry {
    long op_seqno;  /* operation sequence number */
    long kvi_id;
    long kvo_id;
    int  state;
    long n_kvi;     /* number of processed kv in kvi */
    long n_kvo;     /* number of generated kv in kvo */
};

/* Checkpoint log file header */
struct kmr_ckpt_log {
    char magic[8];      /* magic number */
    char info[4];       /* misc info (ex. compressed etc.) */
    int  mode;          /* checkpoint mode */
    int  nprocs;        /* MPI communicator size */
    int  rank;          /* MPI rank */
    struct kmr_ckpt_log_entry data[1];
};

/* Checkpoint file types */
enum kmr_ckpt_type {
    KMR_CKPT_DATA,
    KMR_CKPT_LOG
};

/* Metadata of a checkpoint data file. */
struct kmr_ckpt_data_file {
    long  kvs_id;
    _Bool checked;                      /* if 1, check of restoring data
                                           has been done */
    _Bool merged;                       /* if 1, the data file will be
                                           merged */
    char  fname[KMR_CKPT_PATHLEN];
    char  dname[KMR_CKPT_DIRLEN];
};

/* Information about a merged target data */
struct kmr_ckpt_merge_source {
    int rank;
    long n_kvi;
    long n_kvo;
    long *done_ikv_lst;                 /* list of processed input kv */
    long done_ikv_lst_size;             /* size of done_ikv_lst */
    long kvi_op_seqno;                  /* op_seqno when used as kvi */
    struct kmr_ckpt_data_file *file;
};

/* A merge of checkpoint data from previous ranks */
struct kmr_ckpt_merge {
    int rank;                      /* rank that merges data */
    long kvs_id;                   /* id of kvs whose data are merged */
    struct kmr_ckpt_merge_source *src_lst; /* merged data list */
    int src_lst_count;                     /* size of src_lst */
};

/* Checkpoint merge context data */
struct kmr_ckpt_merge_ctx {
    int max_each_merge;            /* maximum number of merged data
                                      of each merge */
    struct kmr_ckpt_merge *merges; /* list of merges */
    int merges_count;              /* size of merges */
};

/* State of a rank in previous run */
struct kmr_ckpt_prev_state {
    int prev_rank;             /* rank of previous run */
    int prev_nprocs;           /* number of processes used in previous run */
    char *ckpt_dir;            /* checkpoint file directory */
    struct kmr_ckpt_data_file *dataflst;   /* checkpoint data files */
    int dataflst_size;                     /* number of dataflst */
};

/* Dummy kvs id for null kvs */
#define KMR_CKPT_DUMMY_ID 0

/*
Copyright (C) 2012-2016 RIKEN AICS
This library is distributed WITHOUT ANY WARRANTY.  This library can be
redistributed and/or modified under the terms of the BSD 2-Clause License.
*/

#endif /* _KMRCKPT_H */