1. Jeff Squyres
  2. mpi3-fortran

Source

mpi3-fortran / ompi / mca / coll / ml / coll_ml_colls.h

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
/*
 * Copyright (c) 2009-2012 Oak Ridge National Laboratory.  All rights reserved.
 * Copyright (c) 2009-2012 Mellanox Technologies.  All rights reserved.
 * $COPYRIGHT$
 *
 * Additional copyrights may follow
 *
 * $HEADER$
 */

#ifndef MCA_COLL_ML_COLLS_H
#define MCA_COLL_ML_COLLS_H

#include "ompi_config.h"
#include "ompi/mca/bcol/bcol.h"

#define COLL_ML_FN_NAME_LEN 256


/* utility information used to coordinate activities, such as resource
 * management between different functions in the hierarchy
 */
struct mca_coll_ml_utility_data_t {

    /* RLG - temp fix  !!!! - really need to remove this, but right now
     do not want to change the signature of the collective primitives to
     use coll_ml_utility_data_t rather than coll_ml_function_t */
    int dummy;

    /* module */
    struct mca_bcol_base_module_t *bcol_module;

    /* */
    int index_in_consecutive_same_bcol_calls;

    /* number of times functions from this bcol are called in order */
    int n_of_this_type_in_a_row;

    /* number of times functions from this module are called
     * in the collective operation. */
    int n_of_this_type_in_collective;
    int index_of_this_type_in_collective;

};
typedef struct mca_coll_ml_utility_data_t mca_coll_ml_utility_data_t;


/* forward declaration */
struct mca_coll_ml_collective_operation_progress_t;
struct mca_coll_ml_task_status_t;

typedef int (* mca_coll_ml_process_op_fn_t)
    (struct mca_coll_ml_collective_operation_progress_t *coll_op);

typedef int (* mca_coll_ml_task_comp_fn_t)
    (struct mca_coll_ml_task_status_t *coll_op);

typedef int (* mca_coll_ml_fragment_launch_fn_t)
    ( struct mca_coll_ml_collective_operation_progress_t *coll_op);

typedef int (* mca_coll_ml_sequential_task_setup_fn_t)
    ( struct mca_coll_ml_collective_operation_progress_t *coll_op);
/* This data structure defines the dependencies for a given
 * compound operation.  We will use this as a basis for implementing
 * collective operations.
 */
struct mca_coll_ml_compound_functions_t {
    /* label */
    char fn_name[COLL_ML_FN_NAME_LEN];

    /* hierarchy level that is used for this bcol */
    int h_level;

    /* the list of functions that make up this task */
    /* coll_bcol_collective_description_t *bcol_function; */
    mca_bcol_base_coll_fn_desc_t *bcol_function;
    /* task completion function for this compound function */
    mca_coll_ml_task_comp_fn_t task_comp_fn;

    /* module specific information that is a constant on a per group
     * basis
     */
    mca_coll_ml_utility_data_t constant_group_data;

    /* number of dependencies to be satified before these function can be
     * started */
    int num_dependencies;

    /*
     * number of notifications to perform on completion.  The assumption
     * is that a counter will be incremented.
     */
    int num_dependent_tasks;

    /*
     * pointers to counters that need be updated.  This assumes
     * an array of tasks is used to describe the ML level
     * collective operation, with these indecies referencing elements
     * in this array.
     */
    int *dependent_task_indecies;

};

typedef struct mca_coll_ml_compound_functions_t mca_coll_ml_compound_functions_t;

/* Forward declaration for operation_description_t */
struct mca_coll_ml_module_t;

enum {
    COLL_ML_GENERAL_TASK_FN,
    COLL_ML_ROOT_TASK_FN,
    COLL_ML_MAX_TASK_FN
};

enum {
    SEQ_TASK_NOT_STARTED,
    SEQ_TASK_PENDING,
    SEQ_TASK_IN_PROG
};

typedef void (*mca_coll_ml_task_setup_fn_t) (struct mca_coll_ml_task_status_t *task_status, int index, struct mca_coll_ml_compound_functions_t *func);

/*
 * Collective operation definition
 */
struct mca_coll_ml_collective_operation_description_t {

    /*
     * Type of collective opeartion - there are two types:
     * 1) sequential progress through the collectives is sufficient
     * 2) general treatment, popping tasks onto execution queus is needed.
     */
    int progress_type;

    struct mca_coll_ml_topology_t *topo_info;

    /*
     * number of functions in collective operation
     */
    int n_fns;

    /*
     * list of functions
     */
    mca_coll_ml_compound_functions_t *component_functions;

    /*
     * array of lists of functions
     */
    mca_coll_ml_compound_functions_t **comp_fn_arr;

    /*
     * indices into the list - fixes a sequential schedule
     */
    int *sch_idx;

    /*
     * Task setup functions, so far we have only 3 - root and non-root
     */
    mca_coll_ml_task_setup_fn_t task_setup_fn[COLL_ML_MAX_TASK_FN];

    /* number of functions are called for bcols need ordering */
    int n_fns_need_ordering;
};
typedef struct mca_coll_ml_collective_operation_description_t
               mca_coll_ml_collective_operation_description_t;

/* Data structure used to track the state of individual bcol
 * functions.  This is used to track dependencies and completion
 * to progress the ML level function correctly.
 *
 * mca_coll_ml_task_status_t will be associated with an
 * mca_coll_ml_collective_operation_progress_t structure for
 * the duration of the lifetime of a communicator.
 * An array of task statuses will be stored with
 * the mca_coll_ml_collective_operation_progress_t data structure, so
 * that the taks status elements do not need to be moved back to
 * a free list before they are re-used.  When the ML level function
 * is complete, all mca_coll_ml_task_status_t are available for
 * re-use.
 */
struct mca_coll_ml_task_status_t{
    /* need to move this between lists to progress this correctly */
    opal_list_item_t item;

    /* number of dependencies satisfied */
    int n_dep_satisfied;

    /* ***************************************************************
     * Pasha:
     * I'm adding to the status: num_dependencies, num_dependent_tasks and
     * dependent_task_indecies. The information originally resided on mca_coll_ml_compound_functions_t.
     * For collective operation with static nature it is not problem.
     * But for Bcast operation, where run time parameters, like root, actually
     * define the dependency. rt prefix mean run-time.
     */

    /* number of dependencies to be satisfied before these function can be
     * started */
    int rt_num_dependencies;

    /*
     * number of notifications to perform on completion.  The assumption
     * is that a counter will be incremented.
     */
    int rt_num_dependent_tasks;

    /*
     * pointers to counters that need be updated.  This assumes
     * an array of tasks is used to describe the ML level
     * collective operation, with these indecies referencing elements
     * in this array.
     */
    int *rt_dependent_task_indecies;
    /*
     *
     * ***************************************************************/

    /* index in collective schedule */
    int my_index_in_coll_schedule;

    /* function pointers */
    mca_bcol_base_coll_fn_desc_t *bcol_fn;

    /* association with a specific collective task - the ML
     * mca_coll_ml_collective_operation_progress_t stores the
     * specific function parameters */
    struct mca_coll_ml_collective_operation_progress_t *ml_coll_operation;

    mca_coll_ml_task_comp_fn_t task_comp_fn;
};
typedef struct mca_coll_ml_task_status_t mca_coll_ml_task_status_t;

typedef enum mca_coll_ml_pending_type_t {
    REQ_OUT_OF_ORDER = 1,
    REQ_OUT_OF_MEMORY = 1 << 1
} mca_coll_ml_pending_type_t;

/* Forward declaration */
struct ml_payload_buffer_desc_t;
/* Data structure used to track ML level collective operation
 * progress.
 */
struct mca_coll_ml_collective_operation_progress_t {
    /* need this to put on a list properly */
    /* Full message information */
    struct full_message_t {
        /* make this a list item */
        ompi_request_t super;
        /* Next expected fragment.
         * It used for controling order of converter unpack operation */
        size_t next_expected_index;
        /* Pointer to last intilized fragment.
         * It used for controling order of converter unpack operation */
        struct mca_coll_ml_collective_operation_progress_t *last_started_frag;
        /* destination data address in user memory */
        void *dest_user_addr;
        /* source data address in user memory */
        void *src_user_addr;
        /* total message size */
        size_t n_bytes_total;
        /* per-process total message size - relevant for operations
         * such as gather and scatter, where each rank has it's
         * own unique data
         */
        size_t n_bytes_per_proc_total;
        size_t max_n_bytes_per_proc_total;
        /* data processes - from a local perspective */
        size_t n_bytes_delivered;
        /* current offset - where to continue with next fragment */
        size_t n_bytes_scheduled;
        /* number of fragments needed to process this message */
        size_t n_fragments;
        /* number of active frags */
        int n_active;
        /* actual pipeline depth */
        int pipeline_depth;
        /* am I the real root of the collective ? */
        bool root;
        /* collective fragment launcher */
        mca_coll_ml_fragment_launch_fn_t fragment_launcher;
        /* is data contingous */
        bool send_data_continguous;
        bool recv_data_continguous;
        /* data type count */
        int64_t send_count;
        int64_t recv_count;
        /* extent of the data types */
        int32_t send_extent;
        int32_t recv_extent;
        /* send data type */
        struct ompi_datatype_t * send_data_type;
        /* needed for non-contigous buffers */
        size_t offset_into_send_buffer;
        /* receive data type */
        struct ompi_datatype_t * recv_data_type;
        /* needed for non-contigous buffers */
        size_t offset_into_recv_buffer;
        /* Convertors for non contigous data */
        opal_convertor_t send_convertor;
        opal_convertor_t recv_convertor;
        /* Will be used by receiver for #bytes calc in the next frag */
        opal_convertor_t dummy_convertor;
        size_t dummy_conv_position;
        /* Size of packed data */
        size_t send_converter_bytes_packed;
        size_t recv_converter_bytes_packed;
        /* In case if ordering is needed: order num for next frag */
        int next_frag_num;
        /* The variable is used by non-blocking memory synchronization code 
         * for caching bank index */
        int bank_index_to_recycle;
    } full_message;

    /* collective operation being progressed */
    mca_coll_ml_collective_operation_description_t *coll_schedule;
    /* */
    mca_coll_ml_process_op_fn_t process_fn;

    mca_coll_base_module_t *coll_module;

    /* If not null , we have to release next fragment */
    struct mca_coll_ml_collective_operation_progress_t *next_to_process_frag;
    /* pointer to previous fragment */
    struct mca_coll_ml_collective_operation_progress_t *prev_frag;
    /* This flag marks that the fragment is pending on the waiting
     * to be processed prior to recycling
     */
    enum mca_coll_ml_pending_type_t pending;

    /* Fragment data */
    struct fragment_data_t {
        /* current buffer pointer - offset (in bytes) into the user data */
        size_t offset_into_user_buffer;
        size_t offset_into_user_buffer_per_proc;

        /* amount of data (in bytes) in this fragment - amount of data
         * actually processed */
        size_t fragment_size;
        size_t per_rank_fragment_size;
        size_t data_type_count_per_frag;

        /* pointer to full message progress data */
        struct full_message_t *message_descriptor;

        /* ML buffer descriptor attached to this buffer */
        struct ml_payload_buffer_desc_t *buffer_desc;

        /* Which collective algorithm */
        int current_coll_op;
    } fragment_data;

    /* specific function parameters */
    /* the assumption is that the variable parameters passed into
     * the ML level function will persist until the collective operation
     * is complete.  For a blocking function this is until the collective
     * function is exited, and for nonblocking collective functions this
     * is until test or wait completes the collective.
     */
    bcol_function_args_t variable_fn_params;

    struct{
        /* current active function - for sequential algorithms */
        int current_active_bcol_fn;

        /* current function status - not started, or in progress.
         * When the routine has completed, the active bcol index is
         * incremented, so no need to keep track of a completed
         * status.
         */
        int current_bcol_status;

        /* use this call back to setup algorithm specific info
           after each level necessary
          */
       mca_coll_ml_sequential_task_setup_fn_t seq_task_setup;

    } sequential_routine;

    struct{
        /*
         * BCOL function status - individual elements will be posted to
         * ml level component queues, as appropriate.
         */
        mca_coll_ml_task_status_t *status_array;

        /* number of completed tasks - need this for collective completion.
         * Resource completion is tracked by each BCOL module .
         */
        int num_tasks_completed;
    } dag_description;
};
typedef struct mca_coll_ml_collective_operation_progress_t
mca_coll_ml_collective_operation_progress_t;
OBJ_CLASS_DECLARATION(mca_coll_ml_collective_operation_progress_t);

#define OP_ML_MODULE(op) ((mca_coll_ml_module_t *)((op)->coll_module))
#define GET_COMM(op) ((OP_ML_MODULE(op))->comm)
#define IS_COLL_SYNCMEM(op) (ML_MEMSYNC == op->fragment_data.current_coll_op)

#define CHECK_AND_RECYCLE(op)                                                   \
do {                                                                            \
    if (0 == (op)->pending) {                                                   \
        /* Caching 2 values that we can't to touch on op after returing it */   \
        /* back to the free list  (free list may release memory on distruct )*/ \
        struct ompi_communicator_t *comm = GET_COMM(op);                        \
        bool is_coll_sync = IS_COLL_SYNCMEM(op);                                \
        assert(&coll_op->full_message !=                                        \
                                    coll_op->fragment_data.message_descriptor); \
        ML_VERBOSE(10, ("Releasing %p", op));                                   \
        OMPI_FREE_LIST_RETURN(&(((mca_coll_ml_module_t *)(op)->coll_module)->   \
                    coll_ml_collective_descriptors),                            \
                (ompi_free_list_item_t *)op);                                   \
        /* Special check for memory synchronization completion */               \
        /* We have to return it first to free list, since the communicator  */  \
        /* release potentially may trigger ML module distraction and having */  \
        /* the element not on the list may cause memory leak.               */  \
        if (OPAL_UNLIKELY(is_coll_sync)) {                                      \
            OBJ_RELEASE(comm);                                                  \
            /* After this point it is UNSAFE to touch ml module */              \
            /* or communicator */                                               \
        }                                                                       \
    }                                                                           \
} while (0)

#define MCA_COLL_ML_SET_ORDER_INFO(coll_progress, num_frags)                      \
do {                                                                              \
    mca_coll_ml_topology_t *topo = (coll_progress)->coll_schedule->topo_info;     \
    bcol_function_args_t *variable_params = &(coll_progress)->variable_fn_params; \
    if (topo->topo_ordering_info.num_bcols_need_ordering > 0) {                   \
        variable_params->order_info.bcols_started = 0;                            \
        variable_params->order_info.order_num =                                   \
                      topo->topo_ordering_info.next_order_num;                    \
        variable_params->order_info.n_fns_need_ordering =                         \
                       (coll_progress)->coll_schedule->n_fns_need_ordering;       \
        topo->topo_ordering_info.next_order_num += num_frags;                     \
        (coll_progress)->fragment_data.message_descriptor->next_frag_num =        \
                                      variable_params->order_info.order_num + 1;  \
    }                                                                             \
} while (0)

#define MCA_COLL_ML_SET_NEW_FRAG_ORDER_INFO(coll_progress)                                    \
do {                                                                                          \
    mca_coll_ml_topology_t *topo = (coll_progress)->coll_schedule->topo_info;                 \
    if (topo->topo_ordering_info.num_bcols_need_ordering > 0) {                               \
        bcol_function_args_t *variable_params = &(coll_progress)->variable_fn_params;         \
        struct fragment_data_t *frag_data = &(coll_progress)->fragment_data;                  \
        variable_params->order_info.bcols_started = 0;                                        \
        variable_params->order_info.order_num = frag_data->message_descriptor->next_frag_num; \
        variable_params->order_info.n_fns_need_ordering =                                     \
                       (coll_progress)->coll_schedule->n_fns_need_ordering;                   \
        frag_data->message_descriptor->next_frag_num++;                                       \
    }                                                                                         \
} while (0)

#define MCA_COLL_ML_SET_SCHEDULE_ORDER_INFO(schedule)                           \
do {                                                                            \
    int i;                                                                      \
    (schedule)->n_fns_need_ordering = 0;                                        \
    for (i = 0; i < (schedule)->n_fns; ++i) {                                   \
        mca_bcol_base_module_t *current_bcol =                                  \
            (schedule)->component_functions[i].constant_group_data.bcol_module; \
        if (current_bcol->bcol_component->need_ordering) {                      \
            (schedule)->n_fns_need_ordering++;                                  \
        }                                                                       \
    }                                                                           \
} while (0)

#endif /* MCA_COLL_ML_COLLS_H */