root/branches/as-branch/src/server/io.sm @ 7826

/* 
 * (C) 2001 Clemson University and The University of Chicago 
 *
 * See COPYING in top-level directory.
 */

/*
 *  PVFS2 server state machine for driving I/O operations (read and write).
 */

#include <string.h>
#include <assert.h>
#include <stdlib.h> /* sson */

#include "server-config.h"
#include "pvfs2-server.h"
#include "pvfs2-attr.h"
#include "pvfs2-request.h"
#include "pint-distribution.h"
#include "pint-request.h"
#include "pvfs2-internal.h"
#include "pint-segpool.h"

#define NUM_OF_PARALLEL_SMS 1
#define BUFFER_SIZE (256*1024) /* 256KB */

%%

machine pvfs2_io_sm
{
    state prelude
    {
        jump pvfs2_prelude_sm;
        success => send_positive_ack;
        default => send_negative_ack;
    }

    state send_positive_ack
    {
        run io_send_ack;
        success => start_pipelining;
        default => release;
    }

    state send_negative_ack
    {
        run io_send_ack;
        default => release;
    }

    state start_pipelining
    {
        pjmp start_pipelining_sm
        {
            success => pvfs2_pipeline_sm;
        }
        success => cleanup_pipeline;
    }

    state cleanup_pipeline
    {
        run cleanup_pipeline_sm;
        success => send_completion_ack;
        default => release;
    }

    state send_completion_ack
    {
        run io_send_completion_ack;
        default => release;
    }

    state release
    {
        run io_release;
        default => cleanup;
    }

    state cleanup
    {
        run io_cleanup;
        default => terminate;
    }
}

%%

/*
 * Function: io_send_ack()
 *
 * Params:   server_op *s_op, 
 *           job_status_s* js_p
 *
 * Pre:      error code has been set in job status for us to
 *           report to client
 *
 * Post:     response has been sent to client
 *            
 * Returns:  int
 *
 * Synopsis: fills in a response to the I/O request, encodes it,
 *           and sends it to the client via BMI.  Note that it may
 *           send either positive or negative acknowledgements.
 *           
 */
static int io_send_ack(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    int err = -PVFS_EIO;
    job_id_t tmp_id;
    struct server_configuration_s *user_opts = get_server_config_struct();
        
    /* this is where we report the file size to the client before
     * starting the I/O transfer, or else report an error if we
     * failed to get the size, or failed for permission reasons
     */
    s_op->resp.status = js_p->error_code;
    s_op->resp.u.io.bstream_size = s_op->ds_attr.u.datafile.b_size;

    err = PINT_encode(&s_op->resp, PINT_ENCODE_RESP, &(s_op->encoded),
                      s_op->addr, s_op->decoded.enc_type);

    gossip_debug(GOSSIP_IO_DEBUG, "%s: error=%d\n", __func__, err);
    if (err < 0)
    {
        gossip_lerr("Server: IO SM: PINT_encode() failure.\n");
        js_p->error_code = err;
        return SM_ACTION_COMPLETE;
    }

    err = job_bmi_send_list(
        s_op->addr, s_op->encoded.buffer_list, s_op->encoded.size_list,
        s_op->encoded.list_count, s_op->encoded.total_size,
        s_op->tag, s_op->encoded.buffer_type, 0, smcb, 0, js_p,
        &tmp_id, server_job_context, user_opts->server_job_bmi_timeout,
        s_op->req->hints);

    return err;
}

/*
 * Function: start_pipelining_sm()
 *
 * Params:   server_op *s_op, 
 *           job_status_s* js_p
 *
 * Pre:      all of the previous steps have succeeded, so that we
 *           are ready to actually perform the I/O
 *
 * Post:     I/O has been carried out
 *            
 * Returns:  int
 *
 * Synopsis: this is the most important part of the state machine.
 *           we setup the multiple read/write state machines 
 *           to carry out the data transfer
 *           
 */
static PINT_sm_action start_pipelining_sm(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    struct server_configuration_s *user_opts = get_server_config_struct();
    struct filesystem_configuration_s *fs_config;
    struct PINT_server_op *pipeline_op;
    int i, ret, dfile_index;
    PINT_segpool_handle_t seg_handle;

    s_op->u.io.parallel_sms = 0;
    s_op->u.io.total_transferred = 0;

    /* we still have the file size stored in the response structure 
     * that we sent in the previous state, other details come from
     * request
     */
    s_op->u.io.file_data.fsize = s_op->resp.u.io.bstream_size;
    s_op->u.io.file_data.dist = s_op->req->u.io.io_dist;
    s_op->u.io.file_data.server_nr = s_op->req->u.io.server_nr;
    s_op->u.io.file_data.server_ct = s_op->req->u.io.server_ct;

    /* on writes, we allow the bstream to be extended at EOF */
    if (s_op->req->u.io.io_type == PVFS_IO_WRITE)
    {
        gossip_debug(GOSSIP_IO_DEBUG, "%s: issuing pipelining to "
                     "write data.\n", __func__);
        s_op->u.io.file_data.extend_flag = 1;
    }
    else
    {
        gossip_debug(GOSSIP_IO_DEBUG, "%s: issuing pipelining to "
                     "read data.\n", __func__);
        s_op->u.io.file_data.extend_flag = 0;
    }

    s_op->u.io.file_req = s_op->req->u.io.file_req;
    s_op->u.io.file_req_offset = s_op->req->u.io.file_req_offset;
    s_op->u.io.mem_req = NULL;
    s_op->u.io.aggregate_size = s_op->req->u.io.aggregate_size;
    gossip_debug(GOSSIP_IO_DEBUG, "%s: tag=%d\n", __func__, s_op->tag);
    s_op->u.io.user_ptr = NULL;
    s_op->u.io.op = s_op->req->u.io.op; /* AS: operation */
    s_op->u.io.datatype = s_op->req->u.io.datatype; /* AS: dtype */
    s_op->u.io.dfile_array  = s_op->req->u.io.dfile_array; /* AD: dfile_array */
    
    gossip_debug(GOSSIP_IO_DEBUG, "server_ct=%d\n", s_op->req->u.io.server_ct);
    gossip_debug(GOSSIP_IO_DEBUG, "fs_id=%d\n", s_op->req->u.io.fs_id);
    for(i=0; i<s_op->req->u.io.server_ct; i++) {
        gossip_debug(GOSSIP_IO_DEBUG, "dfile_array[%d]=%llu\n", i, 
                     llu(s_op->u.io.dfile_array[i]));
        if(s_op->u.io.dfile_array[i] == s_op->req->u.io.handle) {
            dfile_index = i;
            gossip_debug(GOSSIP_IO_DEBUG, "dfile_index=%d\n", i);
        }
    }

    fs_config = PINT_config_find_fs_id(user_opts, s_op->req->u.io.fs_id);
    if(fs_config)
    {
        /* pick up any buffer settings overrides from fs conf */
        s_op->u.io.buffer_size = fs_config->fp_buffer_size;
        //s_op->u.io.num_of_buffers = fs_config->fp_buffers_per_flow;
        s_op->u.io.num_of_buffers = NUM_OF_PARALLEL_SMS; /* FIXME */
    }

    gossip_debug(GOSSIP_IO_DEBUG, "%s: fsize: %lld, " 
                 "server_nr: %d, server_ct: %d\n", __func__,
        lld(s_op->u.io.file_data.fsize),
        (int)s_op->u.io.file_data.server_nr,
        (int)s_op->u.io.file_data.server_ct);

    gossip_debug(GOSSIP_IO_DEBUG, "      file_req_offset: %lld, "
        "aggregate_size: %lld, handle: %llu\n", 
        lld(s_op->u.io.file_req_offset),
        lld(s_op->u.io.aggregate_size),
        llu(s_op->req->u.io.handle));

    /* setup the request processing states */
    gossip_debug(GOSSIP_IO_DEBUG, "%s: aggregate_size=%lld\n", __func__,
                 lld(s_op->u.io.aggregate_size));
    gossip_debug(GOSSIP_IO_DEBUG, "%s: file_data.fsize=%lld\n", __func__,
                 lld(s_op->u.io.file_data.fsize));

    if(s_op->u.io.buffer_size < 1)
        s_op->u.io.buffer_size = BUFFER_SIZE;
    if(s_op->u.io.num_of_buffers < 1)
        s_op->u.io.num_of_buffers = NUM_OF_PARALLEL_SMS;

#if 0
    /* figure out if the fs config has trove data sync turned on or off
         */
    server_config = get_server_config_struct();
    if(!server_config) {
        js_p->error_code = -PVFS_EINVAL;
        return SM_ACTION_COMPLETE;
    }
    
    fs_config = PINT_config_find_fs_id(server_config, 
                                       s_op->u.pipeline.fs_id);
    if(!fs_config) {
        gossip_err("%s: Failed to get filesystem "
                   "config from fs_id of: %d\n", __func__,
                   s_op->u.pipeline.fs_id);
        js_p->error_code = -PVFS_EINVAL;
        return SM_ACTION_COMPLETE;
    }
#endif

    PINT_segpool_unit_id id[s_op->u.io.num_of_buffers];

    PINT_dist_lookup(s_op->u.io.file_data.dist);
    ret = PINT_segpool_init(s_op->u.io.mem_req,
                            s_op->u.io.file_req,
                            s_op->u.io.file_data.fsize,
                            s_op->u.io.file_req_offset,
                            s_op->u.io.aggregate_size,
                            s_op->u.io.file_data.server_nr,
                            s_op->u.io.file_data.server_ct,
                            s_op->u.io.file_data.dist, /* dist */
                            (s_op->req->u.io.io_type == PVFS_IO_READ ? 
                             PINT_SP_SERVER_READ: PINT_SP_SERVER_WRITE),
                            &seg_handle);

    s_op->u.io.seg_handle = seg_handle;
    gen_mutex_init(&s_op->u.io.mutex); /* FIXME: */
    for(i=0; i<s_op->u.io.num_of_buffers; i++) {
        PINT_segpool_register(seg_handle, &id[i]);
        pipeline_op = malloc(sizeof(*pipeline_op));
        memset(pipeline_op, 0, sizeof(*pipeline_op));

        pipeline_op->u.pipeline.address = s_op->addr;
        pipeline_op->u.pipeline.handle = s_op->req->u.io.handle;
        pipeline_op->u.pipeline.fs_id = s_op->req->u.io.fs_id;

        pipeline_op->u.pipeline.seg_handle = seg_handle;
        pipeline_op->u.pipeline.id = id[i];
        //pipeline_op->u.pipeline.parent = s_op;
        pipeline_op->u.pipeline.hints = s_op->req->hints;
        pipeline_op->u.pipeline.tag = s_op->tag;
        pipeline_op->u.pipeline.buffer_size = BUFFER_SIZE;
        pipeline_op->u.pipeline.io_type = s_op->req->u.io.io_type;

        pipeline_op->u.pipeline.dfile_index = dfile_index;
        pipeline_op->u.pipeline.dfile_count = s_op->u.io.file_data.server_ct;
        pipeline_op->u.pipeline.dist = s_op->u.io.file_data.dist;

        /* figure out if the fs config has trove data sync turned on or off
         */
        pipeline_op->u.pipeline.trove_sync_flag = 
            (fs_config->trove_sync_data ? TROVE_SYNC : 0);
        
        if(s_op->req->u.io.io_type == PVFS_IO_READ) {
            if(!pipeline_op->u.pipeline.buffer) {
                /* if the buffer has not been used, allocate a buffer */
                pipeline_op->u.pipeline.buffer = 
                    BMI_memalloc(pipeline_op->u.pipeline.address,
                                 pipeline_op->u.pipeline.buffer_size, 
                                 BMI_SEND);
                /* TODO: error handling */
                assert(pipeline_op->u.pipeline.buffer);
            }

            gossip_debug(GOSSIP_IO_DEBUG, "smcb->base_frame=%d, frame_count=%d\n", smcb->base_frame, smcb->frame_count);
            ret = PINT_sm_push_frame(smcb, 0, pipeline_op);
            gossip_debug(GOSSIP_IO_DEBUG, "smcb->base_frame=%d, frame_count=%d\n", smcb->base_frame, smcb->frame_count);
            if(ret < 0) {
                js_p->error_code = -PVFS_ENOMEM;
                gossip_debug(GOSSIP_IO_DEBUG, "io: failed to setup nested sm for handle: %llu\n", 
                             llu(s_op->req->u.io.handle));
                return SM_ACTION_COMPLETE;
            }
            s_op->u.io.parallel_sms++;
            gossip_debug(GOSSIP_IO_DEBUG, "%s: parallel_sms=%d\n", 
                         __func__, s_op->u.io.parallel_sms);
        }
        else if(s_op->req->u.io.io_type == PVFS_IO_WRITE) {
            if(!pipeline_op->u.pipeline.buffer){
                /* if the buffer has not been used, allocate a buffer */
                pipeline_op->u.pipeline.buffer = 
                    BMI_memalloc(pipeline_op->u.pipeline.address,
                                 pipeline_op->u.pipeline.buffer_size, 
                                 BMI_RECV);
                /* TODO: error handling */
                assert(pipeline_op->u.pipeline.buffer);
            }

            gossip_debug(GOSSIP_IO_DEBUG, "smcb->base_frame=%d, frame_count=%d\n", smcb->base_frame, smcb->frame_count);
            ret = PINT_sm_push_frame(smcb, 0, pipeline_op);
            if(ret < 0) {
                js_p->error_code = -PVFS_ENOMEM;
                gossip_debug(GOSSIP_IO_DEBUG, "io: failed to setup nested sm for handle: %llu\n", 
                             llu(s_op->req->u.io.handle));
                return SM_ACTION_COMPLETE;
            }
            gossip_debug(GOSSIP_IO_DEBUG, "smcb->base_frame=%d, frame_count=%d\n", smcb->base_frame, smcb->frame_count);
            s_op->u.io.parallel_sms++;
            gossip_debug(GOSSIP_IO_DEBUG, "%s: parallel_sms=%d\n", 
                         __func__, s_op->u.io.parallel_sms);
        }
    }

    js_p->error_code = 0;
    return SM_ACTION_COMPLETE;
}

static PINT_sm_action cleanup_pipeline_sm(struct PINT_smcb *smcb, 
                                          job_status_s *js_p)
{
    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    int i;
    struct PINT_server_op *pipeline_op;
    int task_id;
    int remaining;
    PVFS_error tmp_err;

    for(i=0; i<s_op->u.io.parallel_sms; i++)
    {
        pipeline_op = PINT_sm_pop_frame(smcb, &task_id, &tmp_err, 
            &remaining);
        gossip_debug(GOSSIP_SERVER_DEBUG, 
                     "io: nested sm returned error code: %d\n", tmp_err);
        if(pipeline_op->u.pipeline.buffer) {
            if(s_op->req->u.io.io_type == PVFS_IO_READ) {
                BMI_memfree(pipeline_op->u.pipeline.address, 
                            pipeline_op->u.pipeline.buffer, 
                            s_op->u.io.buffer_size, BMI_SEND);
            }
            else if(s_op->req->u.io.io_type == PVFS_IO_WRITE) {
                BMI_memfree(pipeline_op->u.pipeline.address, 
                            pipeline_op->u.pipeline.buffer, 
                            s_op->u.io.buffer_size, BMI_RECV);
            }
        }
        free(pipeline_op);
    }

    js_p->error_code = 0;
    return SM_ACTION_COMPLETE;
}

                                         
/*
 * Function: io_release()
 *
 * Params:   server_op *b, 
 *           job_status_s* js_p
 *
 * Pre:      we are done with all steps necessary to service
 *           request
 *
 * Post:     operation has been released from the scheduler
 *
 * Returns:  int
 *
 * Synopsis: releases the operation from the scheduler
 */
static PINT_sm_action io_release(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    int ret = 0;
    job_id_t i;

    /*
      tell the scheduler that we are done with this operation (if it
      was scheduled in the first place)
    */
    ret = job_req_sched_release(
        s_op->scheduled_id, smcb, 0, js_p, &i, server_job_context);
    return ret;
}

/*
 * Function: io_cleanup()
 *
 * Params:   server_op *b, 
 *           job_status_s* js_p
 *
 * Pre:      all jobs done, simply need to clean up
 *
 * Post:     everything is free
 *
 * Returns:  int
 *
 * Synopsis: free up any buffers associated with the operation,
 *           including any encoded or decoded protocol structures
 */
static PINT_sm_action io_cleanup(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    char status_string[64] = {0};

    PVFS_strerror_r(s_op->resp.status, status_string, 64);
    PINT_ACCESS_DEBUG(s_op, GOSSIP_ACCESS_DEBUG, "finish (%s)\n", status_string);

    if (s_op->u.io.seg_handle) {
        gossip_debug(GOSSIP_IO_DEBUG, "%s: freeing seg_handle\n", __func__);
        PINT_segpool_destroy(s_op->u.io.seg_handle);
    }
#if 0
    if(s_op->u.io.tmp_buffer)
        free(s_op->u.io.tmp_buffer); /* FIXME */
#endif
    gen_mutex_destroy(&s_op->u.io.mutex); /* FIXME: */

    /* let go of our encoded response buffer, if we appear to have
     * made one
     */
    if (s_op->encoded.total_size)
    {
        PINT_encode_release(&s_op->encoded, PINT_ENCODE_RESP);
    }

    return(server_state_machine_complete(smcb));
}

/*
 * Function: io_send_completion_ack()
 *
 * Params:   server_op *s_op, 
 *           job_status_s* js_p
 *
 * Pre:      IO is completed so that we can report its status
 *
 * Post:     if this is a write, response has been sent to client
 *           if this is a read, do nothing
 *            
 * Returns:  int
 *
 * Synopsis: fills in a response to the I/O request, encodes it,
 *           and sends it to the client via BMI.  Note that it may
 *           send either positive or negative acknowledgements.
 *           
 */
static PINT_sm_action io_send_completion_ack(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    int err = -PVFS_EIO;
    job_id_t tmp_id;
    struct server_configuration_s *user_opts = get_server_config_struct();

    gossip_debug(GOSSIP_IO_DEBUG, "%s: s_op->u.io.parallel_sms=%d\n",
                 __func__, s_op->u.io.parallel_sms);
    /* we only send this trailing ack if we are working on a write
     * operation; otherwise just cut out early
     */
    if (s_op->req->u.io.io_type == PVFS_IO_READ)
    {
        /* normal READ (i.e., without op and datatype) */
        if(s_op->req->u.io.op == 0) { /* AS */
            gossip_debug(GOSSIP_IO_DEBUG, "%s: normal READ\n", __func__);
            js_p->error_code = 0;
            return SM_ACTION_COMPLETE;
        }
        gossip_debug(GOSSIP_IO_DEBUG, "%s: IO_READ with op\n", __func__);
    }

    /* release encoding of the first ack that we sent */
    PINT_encode_release(&s_op->encoded, PINT_ENCODE_RESP);

    /* zero size for safety */
    s_op->encoded.total_size = 0;

    if(s_op->req->u.io.io_type == PVFS_IO_READ) { /* AS */
        switch(s_op->req->u.io.datatype) {
        case (int)0x4c000405: /* MPI_INT */
            s_op->encoded.total_size = sizeof(int); /* FIXME */
            break;
        case (int)0x4c00080b:
            s_op->encoded.total_size = sizeof(double); /* FIXME */
            break;
        default:
            break;
        }
    }

    /*
      fill in response -- status field is the only generic one we
      should have to set
    */
    if(s_op->req->u.io.op != 0 && s_op->req->u.io.io_type == PVFS_IO_READ) /* AS */
        s_op->resp.op = PVFS_SERV_READ_COMPLETION;  /* AS: read with op */
    else
        s_op->resp.op = PVFS_SERV_WRITE_COMPLETION;  /* not IO */
    s_op->resp.status = js_p->error_code;
    if(s_op->req->u.io.io_type == PVFS_IO_READ) { /* AS: read with op */
        s_op->resp.u.read_completion.total_completed =
            s_op->u.io.total_transferred;
        gossip_debug(GOSSIP_IO_DEBUG, "total_transferred=%lld\n", lld(s_op->u.io.total_transferred)); /* AS */
        switch(s_op->req->u.io.datatype) {
        case (int)0x4c000405: /* MPI_INT */
            {
                int *tmp;
                tmp = s_op->u.io.tmp_buffer;
                gossip_debug(GOSSIP_IO_DEBUG, "io_send_completion_ack(), result (INT) to send=%d, s_op->resp.u.read_completion.total_completed=%lld\n", *tmp, lld(s_op->resp.u.read_completion.total_completed));
                s_op->resp.u.read_completion.result.buffer_sz = sizeof(int);
                s_op->resp.u.read_completion.result.buffer = s_op->u.io.tmp_buffer;
                break;
            }
        case (int)0x4c00080b:
            {
                double *tmp;
                tmp = s_op->u.io.tmp_buffer;
                gossip_debug(GOSSIP_IO_DEBUG, "io_send_completion_ack(), result (DOUBLE) to send=%lf, s_op->resp.u.read_completion.total_completed=%lld\n", *tmp, lld(s_op->resp.u.read_completion.total_completed));
                s_op->resp.u.read_completion.result.buffer_sz = sizeof(double);
                s_op->resp.u.read_completion.result.buffer = s_op->u.io.tmp_buffer;
                break;
            }
        }
    }
    else /* AS */
        s_op->resp.u.write_completion.total_completed = 
            s_op->u.io.total_transferred; /* AS */

    gossip_debug(GOSSIP_IO_DEBUG, "%s: total_transferred=%lld\n", __func__,
    lld(s_op->u.io.total_transferred)); err = PINT_encode(
        &s_op->resp, PINT_ENCODE_RESP, &(s_op->encoded),
        s_op->addr, s_op->decoded.enc_type);

    gossip_debug(GOSSIP_IO_DEBUG, "%s: err=%d\n", __func__, err); // err = 0
    if (err < 0)
    {
        gossip_lerr("Server: IO SM: PINT_encode() failure.\n");
        js_p->error_code = err;
        return SM_ACTION_COMPLETE;
    }

    err = job_bmi_send_list(
        s_op->addr, s_op->encoded.buffer_list, s_op->encoded.size_list,
        s_op->encoded.list_count, s_op->encoded.total_size, s_op->tag,
        s_op->encoded.buffer_type, 0, smcb, 0, js_p, &tmp_id,
        server_job_context, user_opts->server_job_bmi_timeout,
        s_op->req->hints);

    gossip_debug(GOSSIP_IO_DEBUG, "%s: err=%d\n", __func__, err); // err = 1

    return err;
}

static enum PINT_server_req_access_type PINT_server_req_access_io(
    struct PVFS_server_req *req)
{
    if(req->u.io.io_type == PVFS_IO_READ)
    {
        return PINT_SERVER_REQ_READONLY;
    }
    return PINT_SERVER_REQ_MODIFY;
}

PINT_GET_OBJECT_REF_DEFINE(io);

struct PINT_server_req_params pvfs2_io_params =
{
    .string_name = "io",
    .perm = PINT_SERVER_CHECK_NONE,
    .access_type = PINT_server_req_access_io,
    .sched_policy = PINT_SERVER_REQ_SCHEDULE,
    .get_object_ref = PINT_get_object_ref_io,
    .state_machine = &pvfs2_io_sm,
};

/*
 * Local variables:
 *  mode: c
 *  c-indent-level: 4
 *  c-basic-offset: 4
 * End:
 *
 * vim: ft=c ts=8 sts=4 sw=4 expandtab
 */