home
***
CD-ROM
|
disk
|
FTP
|
other
***
search
/
ftp.parl.clemson.edu
/
2015-02-07.ftp.parl.clemson.edu.tar
/
ftp.parl.clemson.edu
/
pub
/
pvfs2
/
orangefs-2.8.3-20110323.tar.gz
/
orangefs-2.8.3-20110323.tar
/
orangefs
/
src
/
server
/
io.sm
< prev
next >
Wrap
Text File
|
2009-09-03
|
13KB
|
434 lines
/*
* (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 "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"
%%
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_flow;
default => release;
}
state send_negative_ack
{
run io_send_ack;
default => release;
}
state start_flow
{
run io_start_flow;
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)
{
gossip_debug(GOSSIP_MIRROR_DEBUG,"Executing io_send_ack (io.sm)....\n");
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;
gossip_debug(GOSSIP_MIRROR_DEBUG,"\tbstream_size:%d\n"
,(int)s_op->resp.u.io.bstream_size);
err = PINT_encode(&s_op->resp, PINT_ENCODE_RESP, &(s_op->encoded),
s_op->addr, s_op->decoded.enc_type);
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: io_start_flow()
*
* 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 flow descriptor and post it in order to
* carry out the data transfer
*
*/
static PINT_sm_action io_start_flow(
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();
struct filesystem_configuration_s *fs_conf;
s_op->u.io.flow_d = PINT_flow_alloc();
if (!s_op->u.io.flow_d)
{
js_p->error_code = -PVFS_ENOMEM;
return SM_ACTION_COMPLETE;
}
s_op->u.io.flow_d->hints = s_op->req->hints;
/* 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.flow_d->file_data.fsize = s_op->resp.u.io.bstream_size;
s_op->u.io.flow_d->file_data.dist = s_op->req->u.io.io_dist;
s_op->u.io.flow_d->file_data.server_nr = s_op->req->u.io.server_nr;
s_op->u.io.flow_d->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, "io_start_flow() issuing flow to "
"write data.\n");
s_op->u.io.flow_d->file_data.extend_flag = 1;
}
else
{
gossip_debug(GOSSIP_IO_DEBUG, "io_start_flow() issuing flow to "
"read data.\n");
s_op->u.io.flow_d->file_data.extend_flag = 0;
}
s_op->u.io.flow_d->file_req = s_op->req->u.io.file_req;
s_op->u.io.flow_d->file_req_offset = s_op->req->u.io.file_req_offset;
s_op->u.io.flow_d->mem_req = NULL;
s_op->u.io.flow_d->aggregate_size = s_op->req->u.io.aggregate_size;
s_op->u.io.flow_d->tag = s_op->tag;
s_op->u.io.flow_d->user_ptr = NULL;
s_op->u.io.flow_d->type = s_op->req->u.io.flow_type;
fs_conf = PINT_config_find_fs_id(user_opts,
s_op->req->u.io.fs_id);
if(fs_conf)
{
/* pick up any buffer settings overrides from fs conf */
s_op->u.io.flow_d->buffer_size = fs_conf->fp_buffer_size;
s_op->u.io.flow_d->buffers_per_flow = fs_conf->fp_buffers_per_flow;
}
gossip_debug(GOSSIP_IO_DEBUG, "flow: fsize: %lld, "
"server_nr: %d, server_ct: %d\n",
lld(s_op->u.io.flow_d->file_data.fsize),
(int)s_op->u.io.flow_d->file_data.server_nr,
(int)s_op->u.io.flow_d->file_data.server_ct);
gossip_debug(GOSSIP_IO_DEBUG, " file_req_offset: %lld, "
"aggregate_size: %lld, handle: %llu\n",
lld(s_op->u.io.flow_d->file_req_offset),
lld(s_op->u.io.flow_d->aggregate_size),
llu(s_op->req->u.io.handle));
/* set endpoints depending on type of io requested */
if (s_op->req->u.io.io_type == PVFS_IO_WRITE)
{
s_op->u.io.flow_d->src.endpoint_id = BMI_ENDPOINT;
s_op->u.io.flow_d->src.u.bmi.address = s_op->addr;
s_op->u.io.flow_d->dest.endpoint_id = TROVE_ENDPOINT;
s_op->u.io.flow_d->dest.u.trove.handle = s_op->req->u.io.handle;
s_op->u.io.flow_d->dest.u.trove.coll_id = s_op->req->u.io.fs_id;
}
else if (s_op->req->u.io.io_type == PVFS_IO_READ)
{
s_op->u.io.flow_d->src.endpoint_id = TROVE_ENDPOINT;
s_op->u.io.flow_d->src.u.trove.handle = s_op->req->u.io.handle;
s_op->u.io.flow_d->src.u.trove.coll_id = s_op->req->u.io.fs_id;
s_op->u.io.flow_d->dest.endpoint_id = BMI_ENDPOINT;
s_op->u.io.flow_d->dest.u.bmi.address = s_op->addr;
}
else
{
gossip_lerr("Server: IO SM: unknown IO type requested.\n");
js_p->error_code = -PVFS_EINVAL;
return SM_ACTION_COMPLETE;
}
gossip_debug(GOSSIP_IO_DEBUG,"\tabout to issue job_flow...\n");
err = job_flow(s_op->u.io.flow_d, smcb, 0, js_p, &tmp_id,
server_job_context, user_opts->server_job_flow_timeout
, s_op->req->hints);
gossip_debug(GOSSIP_IO_DEBUG,"\treturn code from job_flow "
"submission:%d\n"
,err);
return err;
}
/*
* 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.flow_d)
{
PINT_flow_free(s_op->u.io.flow_d);
}
/* 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: flow 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,"Executing io_send_completion_ack.\n");
/* 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)
{
js_p->error_code = 0;
return SM_ACTION_COMPLETE;
}
/* 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;
/*
fill in response -- status field is the only generic one we
should have to set
*/
s_op->resp.op = PVFS_SERV_WRITE_COMPLETION; /* not IO */
s_op->resp.status = js_p->error_code;
s_op->resp.u.write_completion.total_completed =
s_op->u.io.flow_d->total_transferred;
err = PINT_encode(
&s_op->resp, PINT_ENCODE_RESP, &(s_op->encoded),
s_op->addr, s_op->decoded.enc_type);
if (err < 0)
{
gossip_lerr("Server: IO SM: PINT_encode() failure.\n");
js_p->error_code = err;
return SM_ACTION_COMPLETE;
}
gossip_debug(GOSSIP_IO_DEBUG,"\ts_op->tag:%d\n",s_op->tag);
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,"return code from sending ack:%d\n"
,err);
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
*/