mirror/hdf5
2
0
Fork 0
mirror of https://github.com/HDFGroup/hdf5.git synced 2024-12-27 08:01:04 +08:00
Code Issues Packages Projects Releases Wiki Activity
3876299d75
hdf5/utils/mirror_vfd/mirror_writer.c
Quincey Koziol 3a21ee0877
Add 'warning density' computation to the warnhist script (#3910) 
* Add 'warning density' computation to the warnhist script, along with several
cleanups to it.   Add "--enable-show-all-warnings" configure (and CMake)
option to disable compiler diagnostic suppression (and therefore show all the
otherwise suppressed compiler diagnostics), disabled by default.  Clean up
a buncn of misc. warnings.

Signed-off-by: Quincey Koziol <qkoziol@amazon.com>
2023-12-29 08:42:35 -06:00

1028 lines
35 KiB
C
Raw Blame History

/* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * *
* Copyright by The HDF Group. *
* All rights reserved. *
* *
* This file is part of HDF5. The full HDF5 copyright notice, including *
* terms governing use, modification, and redistribution, is contained in *
* the COPYING file, which can be found at the root of the source code *
* distribution tree, or in https://www.hdfgroup.org/licenses. *
* If you do not have access to either file, you may request a copy from *
* help@hdfgroup.org. *
* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */
/*
* Remote writer process for the mirror (socket) VFD.
*
* Writer is started with arguments for slaved port.
* Awaits a connection on the socket.
* Handles instructions from the master 'Driver' process.
*
* Current implementation uses Sec2 as the underlying driver when opening a
* file. This is reflected in the source (H5FDmirror.c) of the Mirror driver.
*/
#include "mirror_remote.h"
#ifdef H5_HAVE_MIRROR_VFD
#define HEXDUMP_XMITS 1 /* Toggle whether to print xmit bytes-blob */
/* in detailed logging */
#define HEXDUMP_WRITEDATA 0 /* Toggle whether to print bytes to write */
/* in detailed logging */
#define MW_SESSION_MAGIC 0x88F36B32u
#define MW_SOCK_COMM_MAGIC 0xDF10A157u
#define MW_OPTS_MAGIC 0x3BA8B462u
/* ---------------------------------------------------------------------------
* Structure: struct mirror_session
*
* Bundle of information used to manage the operation of this remote Writer
* in a "session" with the Driver process.
*
* magic (uint32_t)
* Semi-unique "magic" number used to sanity-check a structure for
* validity. MUST equal MW_SESSION_MAGIC to be valid.
*
* sockfd (int)
* File descriptor to the socket.
* Used for receiving bytes from and writing bytes to the Driver
* across the network.
* If not NULL, should be a valid descriptor.
*
* token (uint32t)
* Number used to help sanity-check received transmission from the Writer.
* Each Driver/Writer pairing should have a semi-unique "token" to help
* guard against commands from the wrong entity.
*
* xmit_count (uint32_t)
* Record of transmissions received from the Driver. While the transmission
* protocol should be trustworthy, this serves as an additional guard.
* Starts a 0 and should be incremented for each one-way transmission.
*
* file (H5FD_t *)
* Virtual File handle for the hdf5 file.
* Set on file open if H5Fopen() is successful. If NULL, it is invalid.
*
* log_verbosity (unsigned int)
* The verbosity level for logging. Should be set to one of the values
* defined at the top of this file.
*
* log_stream (FILE *)
* File pointer to which logging output is written. Starts (and ends)
* with a default stream, such as stdout, but can be overridden at
* runtime.
*
* reply (H5FD_mirror_xmit_reply_t)
* Structure space for persistent reply data.
* Should be initialized with basic header info (magic, version, op),
* then with session info (token, xmit count), and finally with specific
* reply info (update xmit_count, status code, and message) before
* transmission.
*
* ----------------------------------------------------------------------------
*/
struct mirror_session {
uint32_t magic;
int sockfd;
uint32_t token;
uint32_t xmit_count;
H5FD_t *file;
loginfo_t *loginfo;
H5FD_mirror_xmit_reply_t reply;
};
/* ---------------------------------------------------------------------------
* Structure: struct sock_comm
*
* Structure for placing the data read and pre-processed from Driver in an
* organized fashion. Useful for pre-processing a received xmit.
*
* magic (uint32_t)
* Semi-unique number to sanity-check structure pointer and validity.
* Must equal MW_SOCK_COMM_MAGIC to be valid.
*
* recd_die (int)
* "Boolean" flag indicating that an explicit shutdown/kill/die command
* was received. Potentially useful for debugging and or "manual"
* operation of the program.
* 0 indicates normal operation, non-0 (1) indicates to die.
*
* xmit_recd (H5FD_mirror_xmit_t *)
* Structure pointer for the "xmit header" as decoded from the raw
* binary stream read from the socket.
*
* raw (char *)
* Pointer to a raw byte array, for storing data as read from the
* socket. Bytes buffer is decoded into xmit_t header and derivative
* structures.
*
* raw_size (size_t)
* Give the size of the `raw` buffer.
*
* ---------------------------------------------------------------------------
*/
struct sock_comm {
uint32_t magic;
int recd_die;
H5FD_mirror_xmit_t *xmit_recd;
char *raw;
size_t raw_size;
};
/* ---------------------------------------------------------------------------
* Structure: struct mirror_writer_opts
*
* Container for default values and options as parsed from the command line.
* Currently rather vestigal, but may be expanded and/or moved to be set by
* Server and passed around as an argument.
*
* magic (uint32_t)
* Semi-unique number to sanity-check structure pointer and validity.
* Must equal MW_OPTS_MAGIC to be valid.
*
* logpath (char *)
* String pointer. Allocated at runtime.
* Specifies file location for logging output.
* May be NULL -- uses default output (e.g., stdout).
*
* ----------------------------------------------------------------------------
*/
struct mirror_writer_opts {
uint32_t magic;
char *logpath;
};
static int do_open(struct mirror_session *session, const H5FD_mirror_xmit_open_t *xmit_open);
/* ---------------------------------------------------------------------------
* Function: session_init
*
* Purpose: Populate mirror_session structure with default and
* options-drived values.
*
* Return: An allocated mirror_session structure pointer on success,
* else NULL.
* ----------------------------------------------------------------------------
*/
static struct mirror_session *
session_init(struct mirror_writer_opts *opts)
{
struct mirror_session *session = NULL;
mirror_log(NULL, V_INFO, "session_init()");
if (NULL == opts || opts->magic != MW_OPTS_MAGIC) {
mirror_log(NULL, V_ERR, "invalid opts pointer");
goto error;
}
session = (struct mirror_session *)malloc(sizeof(struct mirror_session));
if (session == NULL) {
mirror_log(NULL, V_ERR, "can't allocate session structure");
goto error;
}
session->magic = MW_SESSION_MAGIC;
session->sockfd = -1;
session->xmit_count = 0;
session->token = 0;
session->file = NULL;
session->reply.pub.magic = H5FD_MIRROR_XMIT_MAGIC;
session->reply.pub.version = H5FD_MIRROR_XMIT_CURR_VERSION;
session->reply.pub.op = H5FD_MIRROR_OP_REPLY;
session->reply.pub.session_token = 0;
memset(session->reply.message, 0, H5FD_MIRROR_STATUS_MESSAGE_MAX);
/* Options-derived population */
session->loginfo = mirror_log_init(opts->logpath, "W- ", MIRROR_LOG_DEFAULT_VERBOSITY);
return session;
error:
if (session) {
free(session);
}
return NULL;
} /* end session_init() */
/* ---------------------------------------------------------------------------
* Function: session_stop
*
* Purpose: Stop and clean up a session.
* Only do this as part of program termination or aborting startup.
*
* Return: 0 on success, or negative sum of errors encountered.
* ----------------------------------------------------------------------------
*/
static int
session_stop(struct mirror_session *session)
{
int ret_value = 0;
assert(session && (session->magic == MW_SESSION_MAGIC));
mirror_log(session->loginfo, V_INFO, "session_stop()");
/* Close HDF5 file if it is still open (probably in error) */
if (session->file) {
mirror_log(session->loginfo, V_WARN, "HDF5 file still open at cleanup");
if (H5FDclose(session->file) < 0) {
mirror_log(session->loginfo, V_ERR, "H5FDclose() during cleanup!");
ret_value--;
}
}
/* Socket will be closed by parent side of server fork after exit */
/* Close custom logging stream */
if (mirror_log_term(session->loginfo) < 0) {
mirror_log(NULL, V_ERR, "Problem closing logging stream");
ret_value--;
}
session->loginfo = NULL;
/* Invalidate and release structure */
session->magic++;
free(session);
return ret_value;
} /* end session_stop() */
/* ---------------------------------------------------------------------------
* Function: session_start
*
* Purpose: Initiate session, open files.
*
* Return: Success: A valid mirror_session pointer which must later be
* cleaned up with session_stop().
* Failure: NULL, after cleaning up after itself.
* ---------------------------------------------------------------------------
*/
static struct mirror_session *
session_start(int socketfd, const H5FD_mirror_xmit_open_t *xmit_open)
{
struct mirror_session *session = NULL;
struct mirror_writer_opts opts;
mirror_log(NULL, V_INFO, "session_start()");
if (false == H5FD_mirror_xmit_is_open(xmit_open)) {
mirror_log(NULL, V_ERR, "invalid OPEN xmit");
return NULL;
}
opts.magic = MW_OPTS_MAGIC;
opts.logpath = NULL;
session = session_init(&opts);
if (NULL == session) {
mirror_log(NULL, V_ERR, "can't instantiate session");
goto error;
}
session->sockfd = socketfd;
if (do_open(session, xmit_open) < 0) {
mirror_log(NULL, V_ERR, "unable to open file");
goto error;
}
return session;
error:
if (session != NULL) {
if (session_stop(session) < 0) {
mirror_log(NULL, V_WARN, "Can't abort session init");
}
session = NULL;
}
return NULL;
}
/* ---------------------------------------------------------------------------
* Function: _xmit_reply
*
* Purpose: Common operations to send a reply xmit through the session.
*
* Return: 0 on success, -1 if error.
* ----------------------------------------------------------------------------
*/
static int
_xmit_reply(struct mirror_session *session)
{
unsigned char xmit_buf[H5FD_MIRROR_XMIT_REPLY_SIZE];
H5FD_mirror_xmit_reply_t *reply = &(session->reply);
assert(session && (session->magic == MW_SESSION_MAGIC));
mirror_log(session->loginfo, V_ALL, "_xmit_reply()");
reply->pub.xmit_count = session->xmit_count++;
if (H5FD_mirror_xmit_encode_reply(xmit_buf, (const H5FD_mirror_xmit_reply_t *)reply) !=
H5FD_MIRROR_XMIT_REPLY_SIZE) {
mirror_log(session->loginfo, V_ERR, "can't encode reply");
return -1;
}
mirror_log(session->loginfo, V_ALL, "reply xmit data\n```");
mirror_log_bytes(session->loginfo, V_ALL, H5FD_MIRROR_XMIT_REPLY_SIZE, (const unsigned char *)xmit_buf);
mirror_log(session->loginfo, V_ALL, "```");
if (HDwrite(session->sockfd, xmit_buf, H5FD_MIRROR_XMIT_REPLY_SIZE) < 0) {
mirror_log(session->loginfo, V_ERR, "can't write reply to Driver");
return -1;
}
return 0;
} /* end _xmit_reply() */
/* ---------------------------------------------------------------------------
* Function: reply_ok
*
* Purpose: Send an OK reply through the session.
*
* Return: 0 on success, -1 if error.
* ---------------------------------------------------------------------------
*/
static int
reply_ok(struct mirror_session *session)
{
H5FD_mirror_xmit_reply_t *reply = &(session->reply);
assert(session && (session->magic == MW_SESSION_MAGIC));
mirror_log(session->loginfo, V_ALL, "reply_ok()");
reply->status = H5FD_MIRROR_STATUS_OK;
memset(reply->message, 0, H5FD_MIRROR_STATUS_MESSAGE_MAX);
return _xmit_reply(session);
} /* end reply_ok() */
/* ---------------------------------------------------------------------------
* Function: reply_error
*
* Purpose: Send an ERROR reply with message through the session.
* Message may be cut short if it would overflow the available
* buffer in the xmit.
*
* Return: 0 on success, -1 if error.
* ---------------------------------------------------------------------------
*/
static int
reply_error(struct mirror_session *session, const char *msg)
{
H5FD_mirror_xmit_reply_t *reply = &(session->reply);
assert(session && (session->magic == MW_SESSION_MAGIC));
mirror_log(session->loginfo, V_ALL, "reply_error(%s)", msg);
reply->status = H5FD_MIRROR_STATUS_ERROR;
snprintf(reply->message, H5FD_MIRROR_STATUS_MESSAGE_MAX - 1, "%s", msg);
return _xmit_reply(session);
} /* end reply_error() */
/* ---------------------------------------------------------------------------
* Function: do_close
*
* Purpose: Handle an CLOSE operation.
*
* Return: 0 on success, -1 if error.
* ---------------------------------------------------------------------------
*/
static int
do_close(struct mirror_session *session)
{
assert(session && (session->magic == MW_SESSION_MAGIC));
mirror_log(session->loginfo, V_INFO, "do_close()");
if (NULL == session->file) {
mirror_log(session->loginfo, V_ERR, "no file to close!");
reply_error(session, "no file to close");
return -1;
}
if (H5FDclose(session->file) < 0) {
mirror_log(session->loginfo, V_ERR, "H5FDclose()");
reply_error(session, "H5FDclose()");
return -1;
}
session->file = NULL;
if (reply_ok(session) < 0) {
mirror_log(session->loginfo, V_ERR, "can't reply");
reply_error(session, "ok reply failed; session contaminated");
return -1;
}
return 0;
} /* end do_close() */
/* ---------------------------------------------------------------------------
* Function: do_lock
*
* Purpose: Handle a LOCK operation.
*
* Return: 0 on success, -1 if error.
* ---------------------------------------------------------------------------
*/
static int
do_lock(struct mirror_session *session, const unsigned char *xmit_buf)
{
size_t decode_ret = 0;
H5FD_mirror_xmit_lock_t xmit_lock;
assert(session && (session->magic == MW_SESSION_MAGIC) && xmit_buf);
mirror_log(session->loginfo, V_INFO, "do_lock()");
decode_ret = H5FD_mirror_xmit_decode_lock(&xmit_lock, xmit_buf);
if (H5FD_MIRROR_XMIT_LOCK_SIZE != decode_ret) {
mirror_log(session->loginfo, V_ERR, "can't decode set-eoa xmit");
reply_error(session, "remote xmit_eoa_t decoding size failure");
return -1;
}
if (!H5FD_mirror_xmit_is_lock(&xmit_lock)) {
mirror_log(session->loginfo, V_ERR, "not a set-eoa xmit");
reply_error(session, "remote xmit_eoa_t decode failure");
return -1;
}
mirror_log(session->loginfo, V_INFO, "lock rw: (%d)", xmit_lock.rw);
if (H5FDlock(session->file, (bool)xmit_lock.rw) < 0) {
mirror_log(session->loginfo, V_ERR, "H5FDlock()");
reply_error(session, "remote H5FDlock() failure");
return -1;
}
if (reply_ok(session) < 0) {
mirror_log(session->loginfo, V_ERR, "can't reply");
reply_error(session, "ok reply failed; session contaminated");
return -1;
}
return 0;
} /* end do_lock() */
/* ---------------------------------------------------------------------------
* Function: do_open
*
* Purpose: Handle an OPEN operation.
*
* Return: 0 on success, -1 if error.
* ---------------------------------------------------------------------------
*/
static int
do_open(struct mirror_session *session, const H5FD_mirror_xmit_open_t *xmit_open)
{
hid_t fapl_id = H5I_INVALID_HID;
unsigned _flags = 0;
haddr_t _maxaddr = HADDR_UNDEF;
assert(session && (session->magic == MW_SESSION_MAGIC) && xmit_open &&
true == H5FD_mirror_xmit_is_open(xmit_open));
mirror_log(session->loginfo, V_INFO, "do_open()");
if (0 != xmit_open->pub.xmit_count) {
mirror_log(session->loginfo, V_ERR, "open with xmit count not zero!");
reply_error(session, "initial transmission count not zero");
goto error;
}
if (0 != session->token) {
mirror_log(session->loginfo, V_ERR, "open with token already set!");
reply_error(session, "initial session token not zero");
goto error;
}
session->xmit_count = 1;
session->token = xmit_open->pub.session_token;
session->reply.pub.session_token = session->token;
_flags = (unsigned)xmit_open->flags;
_maxaddr = (haddr_t)xmit_open->maxaddr;
/* Check whether the native size_t on the remote machine (Driver) is larger
* than that on the local machine; if so, issue a warning.
* The blob is always an 8-byte bitfield -- check its contents.
*/
if (xmit_open->size_t_blob > (uint64_t)((size_t)(-1))) {
mirror_log(session->loginfo, V_WARN, "Driver size_t is larger than our own");
}
mirror_log(session->loginfo, V_INFO, "to open file %s (flags %d) (maxaddr %d)", xmit_open->filename,
_flags, _maxaddr);
/* Explicitly use Sec2 as the underlying driver for now.
*/
fapl_id = H5Pcreate(H5P_FILE_ACCESS);
if (fapl_id < 0) {
mirror_log(session->loginfo, V_ERR, "can't create FAPL");
reply_error(session, "H5Pcreate() failure");
goto error;
}
if (H5Pset_fapl_sec2(fapl_id) < 0) {
mirror_log(session->loginfo, V_ERR, "can't set FAPL as Sec2");
reply_error(session, "H5Pset_fapl_sec2() failure");
goto error;
}
session->file = H5FDopen(xmit_open->filename, _flags, fapl_id, _maxaddr);
if (NULL == session->file) {
mirror_log(session->loginfo, V_ERR, "H5FDopen()");
reply_error(session, "remote H5FDopen() failure");
goto error;
}
/* FAPL is set and in use; clean up */
if (H5Pclose(fapl_id) < 0) {
mirror_log(session->loginfo, V_ERR, "can't set FAPL as Sec2");
reply_error(session, "H5Pset_fapl_sec2() failure");
goto error;
}
if (reply_ok(session) < 0) {
mirror_log(session->loginfo, V_ERR, "can't reply");
reply_error(session, "ok reply failed; session contaminated");
return -1;
}
return 0;
error:
if (fapl_id > 0) {
H5E_BEGIN_TRY
{
(void)H5Pclose(fapl_id);
}
H5E_END_TRY
}
return -1;
} /* end do_open() */
/* ---------------------------------------------------------------------------
* Function: do_set_eoa
*
* Purpose: Handle a SET_EOA operation.
*
* Return: 0 on success, -1 if error.
* ---------------------------------------------------------------------------
*/
static int
do_set_eoa(struct mirror_session *session, const unsigned char *xmit_buf)
{
size_t decode_ret = 0;
H5FD_mirror_xmit_eoa_t xmit_seoa;
assert(session && (session->magic == MW_SESSION_MAGIC) && xmit_buf);
mirror_log(session->loginfo, V_INFO, "do_set_eoa()");
decode_ret = H5FD_mirror_xmit_decode_set_eoa(&xmit_seoa, xmit_buf);
if (H5FD_MIRROR_XMIT_EOA_SIZE != decode_ret) {
mirror_log(session->loginfo, V_ERR, "can't decode set-eoa xmit");
reply_error(session, "remote xmit_eoa_t decoding size failure");
return -1;
}
if (!H5FD_mirror_xmit_is_set_eoa(&xmit_seoa)) {
mirror_log(session->loginfo, V_ERR, "not a set-eoa xmit");
reply_error(session, "remote xmit_eoa_t decode failure");
return -1;
}
mirror_log(session->loginfo, V_INFO, "set EOA addr %d", xmit_seoa.eoa_addr);
if (H5FDset_eoa(session->file, (H5FD_mem_t)xmit_seoa.type, (haddr_t)xmit_seoa.eoa_addr) < 0) {
mirror_log(session->loginfo, V_ERR, "H5FDset_eoa()");
reply_error(session, "remote H5FDset_eoa() failure");
return -1;
}
if (reply_ok(session) < 0) {
mirror_log(session->loginfo, V_ERR, "can't reply");
reply_error(session, "ok reply failed; session contaminated");
return -1;
}
return 0;
} /* end do_set_eoa() */
/* ---------------------------------------------------------------------------
* Function: do_truncate
*
* Purpose: Handle a TRUNCATE operation.
*
* Return: 0 on success, -1 if error.
* ---------------------------------------------------------------------------
*/
static int
do_truncate(struct mirror_session *session)
{
assert(session && (session->magic == MW_SESSION_MAGIC));
mirror_log(session->loginfo, V_INFO, "do_truncate()");
/* default DXPL ID (0), 0 for "false" closing -- both probably unused */
if (H5FDtruncate(session->file, 0, 0) < 0) {
mirror_log(session->loginfo, V_ERR, "H5FDtruncate()");
reply_error(session, "remote H5FDtruncate() failure");
return -1;
}
if (reply_ok(session) < 0) {
mirror_log(session->loginfo, V_ERR, "can't reply");
reply_error(session, "ok reply failed; session contaminated");
return -1;
}
return 0;
} /* end do_truncate() */
/* ---------------------------------------------------------------------------
* Function: do_unlock
*
* Purpose: Handle an UNLOCK operation.
*
* Return: 0 on success, -1 if error.
* ---------------------------------------------------------------------------
*/
static int
do_unlock(struct mirror_session *session)
{
assert(session && (session->magic == MW_SESSION_MAGIC));
mirror_log(session->loginfo, V_INFO, "do_unlock()");
if (H5FDunlock(session->file) < 0) {
mirror_log(session->loginfo, V_ERR, "H5FDunlock()");
reply_error(session, "remote H5FDunlock() failure");
return -1;
}
if (reply_ok(session) < 0) {
mirror_log(session->loginfo, V_ERR, "can't reply");
reply_error(session, "ok reply failed; session contaminated");
return -1;
}
return 0;
} /* end do_unlock() */
/* ---------------------------------------------------------------------------
* Function: do_write
*
* Purpose: Handle a WRITE operation.
* Receives command, replies; receives & writes data, replies.
*
* It is known that this results in suboptimal performance,
* but handling both small and very, very large write buffers
* with a single "over the wire" exchange
* poses design challenges not worth tackling as of March 2020.
*
* Return: 0 on success, -1 if error.
* ---------------------------------------------------------------------------
*/
static int
do_write(struct mirror_session *session, const unsigned char *xmit_buf)
{
size_t decode_ret = 0;
haddr_t addr = 0;
haddr_t sum_bytes_written = 0;
H5FD_mem_t type = 0;
char *buf = NULL;
ssize_t nbytes_in_packet = 0;
H5FD_mirror_xmit_write_t xmit_write;
assert(session && (session->magic == MW_SESSION_MAGIC) && xmit_buf);
mirror_log(session->loginfo, V_INFO, "do_write()");
if (NULL == session->file) {
mirror_log(session->loginfo, V_ERR, "no open file!");
reply_error(session, "no file open on remote");
return -1;
}
decode_ret = H5FD_mirror_xmit_decode_write(&xmit_write, xmit_buf);
if (H5FD_MIRROR_XMIT_WRITE_SIZE != decode_ret) {
mirror_log(session->loginfo, V_ERR, "can't decode write xmit");
reply_error(session, "remote xmit_write_t decoding size failure");
return -1;
}
if (!H5FD_mirror_xmit_is_write(&xmit_write)) {
mirror_log(session->loginfo, V_ERR, "not a write xmit");
reply_error(session, "remote xmit_write_t decode failure");
return -1;
}
addr = (haddr_t)xmit_write.offset;
type = (H5FD_mem_t)xmit_write.type;
/* Allocate the buffer once -- reuse between loops.
*/
buf = (char *)malloc(sizeof(char) * H5FD_MIRROR_DATA_BUFFER_MAX);
if (NULL == buf) {
mirror_log(session->loginfo, V_ERR, "can't allocate databuffer");
reply_error(session, "can't allocate buffer for receiving data");
return -1;
}
/* got write signal; ready for data */
if (reply_ok(session) < 0) {
mirror_log(session->loginfo, V_ERR, "can't reply");
reply_error(session, "ok reply failed; session contaminated");
return -1;
}
mirror_log(session->loginfo, V_INFO, "to write %zu bytes at %zu", xmit_write.size, addr);
/* The given write may be:
* 1. larger than the allowed single buffer size
* 2. larger than the native size_t of this system
*
* Handle all cases by looping, ingesting as much of the stream as possible
* and writing that part to the file.
*/
sum_bytes_written = 0;
do {
if ((nbytes_in_packet = HDread(session->sockfd, buf, H5FD_MIRROR_DATA_BUFFER_MAX)) < 0) {
mirror_log(session->loginfo, V_ERR, "can't read into databuffer");
reply_error(session, "can't read data buffer");
return -1;
}
mirror_log(session->loginfo, V_INFO, "received %zd bytes", nbytes_in_packet);
if (HEXDUMP_WRITEDATA) {
mirror_log(session->loginfo, V_ALL, "DATA:\n```");
mirror_log_bytes(session->loginfo, V_ALL, (size_t)nbytes_in_packet, (const unsigned char *)buf);
mirror_log(session->loginfo, V_ALL, "```");
}
mirror_log(session->loginfo, V_INFO, "writing %zd bytes at %zu", nbytes_in_packet,
(addr + sum_bytes_written));
if (H5FDwrite(session->file, type, H5P_DEFAULT, (addr + sum_bytes_written), (size_t)nbytes_in_packet,
buf) < 0) {
mirror_log(session->loginfo, V_ERR, "H5FDwrite()");
reply_error(session, "remote H5FDwrite() failure");
return -1;
}
sum_bytes_written += (haddr_t)nbytes_in_packet;
} while (sum_bytes_written < xmit_write.size); /* end while ingesting */
free(buf);
/* signal that we're done here and a-ok */
if (reply_ok(session) < 0) {
mirror_log(session->loginfo, V_ERR, "can't reply");
reply_error(session, "ok reply failed; session contaminated");
return -1;
}
return 0;
} /* end do_write() */
/* ---------------------------------------------------------------------------
* Function: receive_communique
*
* Purpose: Accept bytes from the socket, check for emergency shutdown, and
* sanity-check received bytes.
* The raw bytes read are stored in the sock_comm structure at
* comm->raw.
* The raw bytes are decoded and a xmit_t (header) struct pointer
* in comm is populated at comm->xmit_recd.
*
* Return: 0 on success, -1 if error.
* ---------------------------------------------------------------------------
*/
static int
receive_communique(struct mirror_session *session, struct sock_comm *comm)
{
ssize_t read_ret = 0;
size_t decode_ret;
H5FD_mirror_xmit_t *X = comm->xmit_recd;
assert((session != NULL) && (session->magic == MW_SESSION_MAGIC) && (comm != NULL) &&
(comm->magic == MW_SOCK_COMM_MAGIC) && (comm->xmit_recd != NULL) && (comm->raw != NULL) &&
(comm->raw_size >= H5FD_MIRROR_XMIT_BUFFER_MAX));
mirror_log(session->loginfo, V_INFO, "receive_communique()");
memset(comm->raw, 0, comm->raw_size);
comm->recd_die = 0;
mirror_log(session->loginfo, V_INFO, "ready to receive"); /* TODO */
if ((read_ret = HDread(session->sockfd, comm->raw, H5FD_MIRROR_XMIT_BUFFER_MAX)) < 0) {
mirror_log(session->loginfo, V_ERR, "read:%zd", read_ret);
goto error;
}
mirror_log(session->loginfo, V_INFO, "received %zd bytes", read_ret);
if (HEXDUMP_XMITS) {
mirror_log(session->loginfo, V_ALL, "```", read_ret);
mirror_log_bytes(session->loginfo, V_ALL, (size_t)read_ret, (const unsigned char *)comm->raw);
mirror_log(session->loginfo, V_ALL, "```");
} /* end if hexdump transmissions received */
/* old-fashioned manual kill (for debugging) */
if (!strncmp("GOODBYE", comm->raw, 7)) {
mirror_log(session->loginfo, V_INFO, "received GOODBYE");
comm->recd_die = 1;
goto done;
}
decode_ret = H5FD_mirror_xmit_decode_header(X, (const unsigned char *)comm->raw);
if (H5FD_MIRROR_XMIT_HEADER_SIZE != decode_ret) {
mirror_log(session->loginfo, V_ERR, "header decode size mismatch: expected (%z), got (%z)",
H5FD_MIRROR_XMIT_HEADER_SIZE, decode_ret);
/* Try to tell Driver that it should stop */
reply_error(session, "xmit size mismatch");
goto error;
}
if (!H5FD_mirror_xmit_is_xmit(X)) {
mirror_log(session->loginfo, V_ERR, "bad magic: 0x%X", X->magic);
/* Try to tell Driver that it should stop */
reply_error(session, "bad magic");
goto error;
}
if (session->xmit_count != X->xmit_count) {
mirror_log(session->loginfo, V_ERR, "xmit_count mismatch exp:%d recd:%d", session->xmit_count,
X->xmit_count);
/* Try to tell Driver that it should stop */
reply_error(session, "xmit_count mismatch");
goto error;
}
if ((session->token > 0) && (session->token != X->session_token)) {
mirror_log(session->loginfo, V_ERR, "wrong session");
/* Try to tell Driver that it should stop */
reply_error(session, "wrong session");
goto error;
}
session->xmit_count++;
done:
return 0;
error:
return -1;
} /* end receive_communique() */
/* ---------------------------------------------------------------------------
* Function: process_instructions
*
* Purpose: Receive and handle all instructions from Driver.
*
* Return: 0 on success, -1 if error.
* ---------------------------------------------------------------------------
*/
static int
process_instructions(struct mirror_session *session)
{
struct sock_comm comm;
char *xmit_buf = NULL; /* raw bytes */
size_t buf_size;
H5FD_mirror_xmit_t xmit_recd; /* for decoded xmit header */
assert(session && (session->magic == MW_SESSION_MAGIC));
mirror_log(session->loginfo, V_INFO, "process_instructions()");
buf_size = H5FD_MIRROR_XMIT_BUFFER_MAX * sizeof(char);
if (NULL == (xmit_buf = malloc(buf_size))) {
mirror_log(session->loginfo, V_ERR, "out of memory");
goto error;
}
comm.magic = MW_SOCK_COMM_MAGIC;
comm.recd_die = 0; /* Flag for program to terminate */
comm.xmit_recd = &xmit_recd;
comm.raw = xmit_buf;
comm.raw_size = buf_size;
while (1) { /* sill-listening infinite loop */
/* Use convenience structure for raw/decoded info in/out */
if (receive_communique(session, &comm) < 0) {
mirror_log(session->loginfo, V_ERR, "problem reading socket");
goto error;
}
if (comm.recd_die) {
goto done;
}
switch (xmit_recd.op) {
case H5FD_MIRROR_OP_CLOSE:
if (do_close(session) < 0) {
goto error;
}
goto done;
case H5FD_MIRROR_OP_LOCK:
if (do_lock(session, (const unsigned char *)xmit_buf) < 0) {
goto error;
}
break;
case H5FD_MIRROR_OP_OPEN:
mirror_log(session->loginfo, V_ERR, "OPEN xmit during session");
reply_error(session, "illegal OPEN xmit during session");
goto error;
case H5FD_MIRROR_OP_SET_EOA:
if (do_set_eoa(session, (const unsigned char *)xmit_buf) < 0) {
goto error;
}
break;
case H5FD_MIRROR_OP_TRUNCATE:
if (do_truncate(session) < 0) {
goto error;
}
break;
case H5FD_MIRROR_OP_UNLOCK:
if (do_unlock(session) < 0) {
goto error;
}
break;
case H5FD_MIRROR_OP_WRITE:
if (do_write(session, (const unsigned char *)xmit_buf) < 0) {
goto error;
}
break;
default:
mirror_log(session->loginfo, V_ERR, "unrecognized transmission");
reply_error(session, "unrecognized transmission");
goto error;
} /* end switch (xmit_recd.op) */
} /* end while still listening */
done:
comm.magic = 0; /* invalidate structure, on principle */
xmit_recd.magic = 0; /* invalidate structure, on principle */
free(xmit_buf);
return 0;
error:
free(xmit_buf);
return -1;
} /* end process_instructions() */
/* ---------------------------------------------------------------------------
* Function: run_writer
*
* Purpose: Initiate Writer operations.
*
* Receives as parameters a socket which has accepted the
* connection to the Driver and the OPEN xmit (which must be
* decoded into the structure and verified prior to being passed
* to this function).
*
* Is not responsible for closing or cleaning up any of the
* received parameters.
*
* Return: Success: SUCCEED
* Failure: FAIL
* ---------------------------------------------------------------------------
*/
herr_t
run_writer(int socketfd, H5FD_mirror_xmit_open_t *xmit_open)
{
struct mirror_session *session = NULL;
int ret_value = SUCCEED;
session = session_start(socketfd, xmit_open);
if (NULL == session) {
mirror_log(NULL, V_ERR, "Can't start session -- aborting");
ret_value = FAIL;
}
else {
if (process_instructions(session) < 0) {
mirror_log(session->loginfo, V_ERR, "problem processing instructions");
ret_value = FAIL;
}
if (session_stop(session) < 0) {
mirror_log(NULL, V_ERR, "Can't stop session -- going down hard");
ret_value = FAIL;
}
}
return ret_value;
} /* end run_writer */
#endif /* H5_HAVE_MIRROR_VFD */
Reference in New Issue View Git Blame Copy Permalink