prepare-write-op.c 2.88 KB
Newer Older
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
/*
 * (C) 2017 The University of Chicago
 * 
 * See COPYRIGHT in top-level directory.
 */
#include "prepare-write-op.h"
#include "write-op-impl.h"
#include "utlist.h"
#include "log.h"

static void prepare_write(uint64_t* cur_offset,
                          wr_action_write_t action,
                          void** ptr,
                          size_t* len)
{
	uint64_t pos = *cur_offset;
	*cur_offset += action->len;
	*ptr         = (void*)action->buffer.as_pointer;
	*len         = action->len;
	action->buffer.as_offset = pos;
}

static void prepare_write_full(uint64_t* cur_offset,
                               wr_action_write_full_t action,
                               void** ptr,
                               size_t* len)
{
	uint64_t pos = *cur_offset;
	*cur_offset += action->len;
	*ptr         = (void*)action->buffer.as_pointer;
	*len         = action->len;
	action->buffer.as_offset = pos;
}

static void prepare_write_same(uint64_t* cur_offset,
                               wr_action_write_same_t action,
                               void** ptr,
                               size_t* len)
{
	uint64_t pos = *cur_offset;
	*cur_offset += action->data_len;
	*ptr         = (void*)action->buffer.as_pointer;
	*len         = action->data_len;
	action->buffer.as_offset = pos;
}

static void prepare_append(uint64_t* cur_offset,
                           wr_action_append_t action,
                           void** ptr,
                           size_t* len)
{
	uint64_t pos = *cur_offset;
	*cur_offset += action->len;
	*ptr         = (void*)action->buffer.as_pointer;
	*len         = action->len;
	action->buffer.as_offset = pos;
}

void prepare_write_op(margo_instance_id mid, mobject_store_write_op_t write_op) 
{
	if(write_op->use_local_pointers == 0) return;
	if(write_op->num_actions == 0) return;	

	wr_action_base_t action;

	void** pointers = (void**)calloc(write_op->num_actions, sizeof(void*));
	size_t* lengths = (size_t*)calloc(write_op->num_actions, sizeof(size_t));
	uint64_t current_offset = 0;
	size_t i = 0;

	DL_FOREACH(write_op->actions, action) {

		switch(action->type) {
		case WRITE_OPCODE_WRITE:
			prepare_write(&current_offset, 
				(wr_action_write_t)action, pointers+i, lengths+i);
			i += 1;
			break;
		case WRITE_OPCODE_WRITE_FULL:
			prepare_write_full(&current_offset,
				(wr_action_write_full_t)action, pointers+i, lengths+i);
			i += 1;
			break;
		case WRITE_OPCODE_WRITE_SAME:
			prepare_write_same(&current_offset, 
				(wr_action_write_same_t)action, pointers+i, lengths+i);
			i += 1;
			break;
		case WRITE_OPCODE_APPEND:
			prepare_append(&current_offset, 
				(wr_action_append_t)action, pointers+i, lengths+i);
			i += 1;
			break;
		}	
	}

	uint32_t count = i;
	if(count != 0) {
		hg_return_t ret = margo_bulk_create(mid, count,
    						pointers, lengths, HG_BULK_READ_ONLY, 
							&(write_op->bulk_handle));
		// TODO handle error

	}

	write_op->use_local_pointers = 0;
}