From 323d1c878e810f04d7cbfe0c8d55cf367460fc91 Mon Sep 17 00:00:00 2001 From: Anton Pegushin Date: Tue, 4 Aug 2020 13:46:55 -0700 Subject: [PATCH] Add write-messages-to-file option. --- src/defs.h | 11 +++++- src/server.cpp | 91 +++++++++++++++++++++++++++++++++++++++++++++++- src/server.h | 45 ++++++++++++++++++++++++ src/sockperf.cpp | 72 ++++++++++++++++++++++++++++++++++++++ 4 files changed, 217 insertions(+), 2 deletions(-) diff --git a/src/defs.h b/src/defs.h index dac45b0c..fdce783c 100644 --- a/src/defs.h +++ b/src/defs.h @@ -138,6 +138,8 @@ const uint32_t TEST_FIRST_CONNECTION_FIRST_PACKET_TTL_THRESHOLD_MSEC = 50; #define MAX_ACTIVE_FD_NUM \ 1024 /* maximum number of active connection to the single TCP addr:port \ */ +#define DEFAULT_BUF_ALIGNMENT 512 +#define DEFAULT_FILE_FLAGS_OPT 0 #ifdef USING_VMA_EXTRA_API #define MAX_VMA_COMPS 1024 /* maximum size for the VMA completions array for VMA Poll */ #endif @@ -221,7 +223,10 @@ enum { OPT_DUMMY_SEND, // 41 OPT_RATE_LIMIT, // 42 OPT_UC_REUSEADDR, // 43 - OPT_FULL_RTT // 44 + OPT_FULL_RTT, // 44 + OPT_WRITE_MSG_FILE_PATH, // 45 + OPT_WRITE_MSG_FILE_FLAGS, // 46 + OPT_WRITE_MSG_BUF_ALIGN // 47 }; static const char *const round_trip_str[] = { "latency", "rtt" }; @@ -679,6 +684,10 @@ struct user_params_t { uint32_t dummy_mps; // client side only TicksDuration dummySendCycleDuration; // client side only uint32_t rate_limit; + bool b_write_msg_to_file; + char write_msg_filepath[MAX_PATH_LENGTH]; + uint32_t write_msg_file_flags_opt; + uint32_t write_msg_buf_alignment; }; struct mutable_params_t {}; diff --git a/src/server.cpp b/src/server.cpp index 62fc6ffa..850a2af3 100644 --- a/src/server.cpp +++ b/src/server.cpp @@ -155,7 +155,11 @@ void ServerBase::cleanupAfterLoop() { //------------------------------------------------------------------------------ template Server::Server(int _fd_min, int _fd_max, int _fd_num) - : ServerBase(m_ioHandler), m_ioHandler(_fd_min, _fd_max, _fd_num) {} + : ServerBase(m_ioHandler), m_ioHandler(_fd_min, _fd_max, _fd_num), + b_write_msg_file_init(false), + write_msg_file(NULL), + write_msg_fd(0), open_fd_default_flags(O_CREAT | O_RDWR | O_TRUNC), + write_msg_buf(NULL), write_msg_buf_aligned(NULL) {} //------------------------------------------------------------------------------ template @@ -318,6 +322,11 @@ int Server::server_accept(int ifd) { inet_ntoa(addr.sin_addr), ntohs(addr.sin_port), active_ifd); } do_accept = true; + + if (s_user_params.b_write_msg_to_file && !b_write_msg_file_init) { + setup_write_msg_to_file(); + } + break; } } @@ -351,11 +360,91 @@ int Server::server_accept(int ifd) { } } } + } else if (g_fds_array[ifd]->sock_type == SOCK_DGRAM) { + if (s_user_params.b_write_msg_to_file && !b_write_msg_file_init) { + setup_write_msg_to_file(); + } } return active_ifd; } +//------------------------------------------------------------------------------ +template +void Server::setup_write_msg_to_file() { + + /* 0 - write to file through a file stream + * 1 - open_fd_default_flags | O_DSYNC + * 2 - open_fd_default_flags | O_DIRECT | O_DSYNC */ + switch (s_user_params.write_msg_file_flags_opt) + { + case 0: + write_msg_file = fopen(s_user_params.write_msg_filepath, "w+"); + if (write_msg_file == NULL) { + log_err("Failed to open file stream for file: %s.", s_user_params.write_msg_filepath); + exit_with_log("Failed to enable writing messages to file. Exiting sockperf.", SOCKPERF_ERR_FATAL); + } + break; + case 1: + open_fd_align_buffer(O_DSYNC); + break; + case 2: + open_fd_align_buffer(O_DSYNC | O_DIRECT); + break; + default: + exit_with_log("Unrecognized file sync option provided. Exiting sockperf.", SOCKPERF_ERR_BAD_ARGUMENT); + break; + } + + b_write_msg_file_init = true; +} + +//------------------------------------------------------------------------------ +template +void Server::cleanup_write_msg_to_file() { + + if (write_msg_file != NULL) { + if (fclose(write_msg_file) != 0) { + log_err("Failed to close file stream for file: %s.", s_user_params.write_msg_filepath); + } + write_msg_file = NULL; + } else { + if (write_msg_fd > 0) { + if (close(write_msg_fd) != 0) { + log_err("Failed to close file descriptor for file: %s.", s_user_params.write_msg_filepath); + } + } + free(write_msg_buf); + } + + b_write_msg_file_init = false; +} + +/* To write to a file opened with O_DIRECT userspace buffer needs to be aligned + * by a multiple of 512 bytes and the buffer length needs to be a multiple of 512 bytes. */ +#define DIRECT_IO_BUF_ALIGN(PTR, ALIGNVAL) (((uintptr_t) (PTR) + ((ALIGNVAL) - 1)) & ~((uintptr_t) ((ALIGNVAL) - 1))) + +//------------------------------------------------------------------------------ +template +void Server::open_fd_align_buffer(int flags) { + + write_msg_fd = open(s_user_params.write_msg_filepath, open_fd_default_flags | flags, S_IRUSR | S_IWUSR); + + if (write_msg_fd > 0) { + /* At server startup actual message sizes used by sockperf client are unknown, + * MAX_PAYLOAD_SIZE is used to ensure client is able to send messages of any supported size. */ + write_msg_buf = (uint8_t*)MALLOC(MAX_PAYLOAD_SIZE + s_user_params.write_msg_buf_alignment); + if (write_msg_buf == NULL) { + exit_with_log("Failed to allocate memory for write_msg_buf.", SOCKPERF_ERR_NO_MEMORY); + } + write_msg_buf_aligned = (uint8_t*)DIRECT_IO_BUF_ALIGN(write_msg_buf, s_user_params.write_msg_buf_alignment); + } else { + log_err("Failed to open file: %s.", s_user_params.write_msg_filepath); + exit_with_log("Failed to enable writing messages to file. Exiting sockperf.", SOCKPERF_ERR_FATAL); + } +} +#undef DIRECT_IO_BUF_ALIGN + //------------------------------------------------------------------------------ template void server_handler(int _fd_min, int _fd_max, int _fd_num) { diff --git a/src/server.h b/src/server.h index 8b887cb9..5e91847a 100644 --- a/src/server.h +++ b/src/server.h @@ -98,6 +98,16 @@ class Server : public ServerBase { int server_accept(int ifd); private: + bool b_write_msg_file_init; + FILE *write_msg_file; + int write_msg_fd; + int open_fd_default_flags; + uint8_t *write_msg_buf; + uint8_t *write_msg_buf_aligned; + void setup_write_msg_to_file(); + void cleanup_write_msg_to_file(); + void open_fd_align_buffer(int flags); + SwitchActivityInfo m_switchActivityInfo; SwitchCalcGaps m_switchCalcGaps; }; @@ -186,6 +196,7 @@ inline bool Server::server_receive_t if (ret == RET_SOCKET_SHUTDOWN) { if (l_fds_ifd->sock_type == SOCK_STREAM) { close_ifd(l_fds_ifd->next_fd, ifd, l_fds_ifd); + cleanup_write_msg_to_file(); } return (do_update); } @@ -223,6 +234,7 @@ inline bool Server::server_receive_t print_log("Message received was larger than expected, message ignored.", l_fds_ifd); close_ifd(l_fds_ifd->next_fd, ifd, l_fds_ifd); + cleanup_write_msg_to_file(); return (do_update); } @@ -255,6 +267,38 @@ inline bool Server::server_receive_t hexdump(m_pMsgReply->getBuf(), MsgHeader::EFFECTIVE_SIZE); #endif /* LOG_TRACE_MSG_IN */ + /* if requested persist each message to file */ + if (s_user_params.b_write_msg_to_file) { + + uint8_t *msg = m_pMsgReply->getBuf(); + int msg_len = m_pMsgReply->getLength(); + + if (write_msg_file) { + assert(s_user_params.write_msg_file_flags_opt == DEFAULT_FILE_FLAGS_OPT); + + size_t wb = fwrite(msg, sizeof(uint8_t), msg_len, write_msg_file); + + if (wb != (size_t)msg_len) { + log_err("Failed to write complete message to file stream - wrote %zu of %d bytes.", wb, msg_len); + exit_with_log("Failed to write message to file stream.", SOCKPERF_ERR_FATAL); + } + + if (fflush(write_msg_file) != 0) { + exit_with_log("Failed to flush file stream.", SOCKPERF_ERR_FATAL); + } + } else { + assert(s_user_params.write_msg_file_flags_opt != DEFAULT_FILE_FLAGS_OPT); + + memcpy(write_msg_buf_aligned, msg, msg_len); + ssize_t wb = write(write_msg_fd, write_msg_buf_aligned, msg_len); + + if (wb != msg_len) { + log_err("Failed to write complete message to file descriptor - wrote %zu of %d bytes.", wb, msg_len); + exit_with_log("Failed to write message to file descriptor.", SOCKPERF_ERR_FATAL); + } + } + } + if (g_b_exit) return (!do_update); if (!m_pMsgReply->isClient()) { /* 6: shift to start of cycle buffer in case receiving buffer is empty and @@ -322,6 +366,7 @@ inline bool Server::server_receive_t if (unlikely(ret == RET_SOCKET_SHUTDOWN)) { if (l_fds_ifd->sock_type == SOCK_STREAM) { close_ifd(l_fds_ifd->next_fd, ifd, l_fds_ifd); + cleanup_write_msg_to_file(); } return (do_update); } diff --git a/src/sockperf.cpp b/src/sockperf.cpp index b619e691..d8038c1e 100644 --- a/src/sockperf.cpp +++ b/src/sockperf.cpp @@ -1358,6 +1358,23 @@ static int proc_mode_server(int id, int argc, const char **argv) { "Set maximum message size that the server can receive bytes (default 65507)." }, { 'g', AOPT_NOARG, aopt_set_literal('g'), aopt_set_string("gap-detection"), "Enable gap-detection." }, + { OPT_WRITE_MSG_FILE_PATH, + AOPT_ARG, + aopt_set_literal('o'), + aopt_set_string("output-file"), + "Specify path to a file for sockperf server to write received messages to." }, + { OPT_WRITE_MSG_FILE_FLAGS, + AOPT_ARG, + aopt_set_literal(0), + aopt_set_string("write-flags"), + "Flags to use to open the file, where received messages will be written to (default 0). " + " Accepted values: 0: No flags, use file stream; 1: D_SYNC; 2: (O_DIRECT | O_DSYNC) " }, + { OPT_WRITE_MSG_BUF_ALIGN, + AOPT_ARG, + aopt_set_literal(0), + aopt_set_string("write-align"), + "Buffer alignment, required for O_DIRECT (default 512). " + "Specified value needs to be a power of two, greater or equal to 512." }, { 0, AOPT_NOARG, aopt_set_literal(0), aopt_set_string(NULL), NULL } }; @@ -1466,6 +1483,57 @@ static int proc_mode_server(int id, int argc, const char **argv) { if (!rc && aopt_check(server_obj, 'g')) { s_user_params.b_server_detect_gaps = true; } + + if (!rc && aopt_check(server_obj, OPT_WRITE_MSG_FILE_PATH)) { + /* Writing messages to a file is currently only supported for a single-threaded sockperf server */ + if (s_user_params.mthread_server || aopt_check(server_obj, OPT_THREADS_NUM)) { + log_msg("Writing messages to file (option '-o ') is not supported for multi-threaded server"); + rc = SOCKPERF_ERR_UNSUPPORTED; + } else { + const char *optarg = aopt_value(server_obj, OPT_WRITE_MSG_FILE_PATH); + if (optarg) { + strncpy(s_user_params.write_msg_filepath, optarg, MAX_ARGV_SIZE); + s_user_params.write_msg_filepath[MAX_PATH_LENGTH - 1] = '\0'; + + s_user_params.b_write_msg_to_file = true; + } else { + log_msg("'-%c' Valid path to output file is required, when writing messages to file", 'o'); + rc = SOCKPERF_ERR_BAD_ARGUMENT; + } + } + } + + if (!rc && aopt_check(server_obj, OPT_WRITE_MSG_FILE_FLAGS)) { + const char *optarg = aopt_value(server_obj, OPT_WRITE_MSG_FILE_FLAGS); + long int value = strtol(optarg, NULL, 0); + + /* 0 - write to file through a file stream + * 1 - open_fd_default_flags | O_DSYNC + * 2 - open_fd_default_flags | O_DIRECT | O_DSYNC */ + if (value < 0 || value > 2) { + log_err("Invalid sync flag value: %s . Setting the value to default: %d", optarg, DEFAULT_FILE_FLAGS_OPT); + s_user_params.write_msg_file_flags_opt = DEFAULT_FILE_FLAGS_OPT; + } else { + s_user_params.write_msg_file_flags_opt = value; + + /* When not using a file stream to write messages to file, check that Direct IO (O_DIRECT) + * requirements for userspace buffer alignment are met. */ + if (value != DEFAULT_FILE_FLAGS_OPT) { + if (aopt_check(server_obj, OPT_WRITE_MSG_BUF_ALIGN)) { + const char *optarg = aopt_value(server_obj, OPT_WRITE_MSG_BUF_ALIGN); + long int alignment = strtol(optarg, NULL, 0); + + if (alignment >= DEFAULT_BUF_ALIGNMENT && ((alignment & (alignment - 1)) == 0)) { + s_user_params.write_msg_buf_alignment = alignment; + } else { + log_msg("Invalid buffer alignment value: %s . Setting the value to default: %d bytes", + optarg, DEFAULT_BUF_ALIGNMENT); + /* The default alignment has already been set at args init, no need to assign again. */ + } + } + } + } + } } if (rc) { @@ -2221,6 +2289,10 @@ void set_defaults() { s_user_params.dummy_mps = 0; memset(s_user_params.feedfile_name, 0, sizeof(s_user_params.feedfile_name)); s_user_params.tos = 0x00; + s_user_params.b_write_msg_to_file = false; + memset(s_user_params.write_msg_filepath, 0, sizeof(s_user_params.write_msg_filepath)); + s_user_params.write_msg_file_flags_opt = 0; + s_user_params.write_msg_buf_alignment = DEFAULT_BUF_ALIGNMENT; } //------------------------------------------------------------------------------