mirrors/pv
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

New option "--direct-io".

Browse Source
This commit is contained in:
Andrew Wood
2023-07-27 21:44:30 +01:00
parent 9f07a28065
commit ae81f83e9b
18 changed files with 262 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/include/options.h
+1
View File
@@ -44,6 +44,7 @@ struct opts_s { /* structure describing run-time options */
unsigned int skip_errors; /* skip read errors counter */
bool stop_at_size; /* set if we stop at "size" bytes */
bool sync_after_write; /* set if we sync after every write */
bool direct_io; /* set if O_DIRECT is to be used */
double interval; /* interval between updates */
double delay_start; /* delay before first display */
unsigned int watch_pid; /* process to watch fds of */
src/include/pv-internal.h
+2
View File
@@ -94,6 +94,8 @@ struct pvstate_s {
unsigned int skip_errors; /* skip read errors counter */
bool stop_at_size; /* set if we stop at "size" bytes */
bool sync_after_write; /* set if we sync after every write */
bool direct_io; /* set if O_DIRECT is to be used */
bool direct_io_changed; /* set when direct_io is changed */
bool no_splice; /* never use splice() */
unsigned long long rate_limit; /* rate limit, in bytes per second */
unsigned long long target_buffer_size; /* buffer size (0=default) */
src/include/pv.h
+1
View File
@@ -107,6 +107,7 @@ extern void pv_state_no_op_set(pvstate_t, bool);
extern void pv_state_skip_errors_set(pvstate_t, unsigned int);
extern void pv_state_stop_at_size_set(pvstate_t, bool);
extern void pv_state_sync_after_write_set(pvstate_t, bool);
extern void pv_state_direct_io_set(pvstate_t, bool);
extern void pv_state_rate_limit_set(pvstate_t, unsigned long long);
extern void pv_state_target_buffer_size_set(pvstate_t, unsigned long long);
extern void pv_state_no_splice_set(pvstate_t, bool);
src/main/help.c
+2
View File
@@ -93,6 +93,8 @@ void display_help(void)
N_("stop after --size bytes have been transferred")},
{"-Y", "--sync", 0,
N_("flush cache to disk after every write")},
{"-K", "--direct-io", 0,
N_("use direct I/O to bypass cache")},
#ifdef HAVE_IPC
{"-R", "--remote", N_("PID"),
N_("update settings of process PID")},
src/main/main.c
+1
View File
@@ -202,6 +202,7 @@ int main(int argc, char **argv)
pv_state_skip_errors_set(state, opts->skip_errors);
pv_state_stop_at_size_set(state, opts->stop_at_size);
pv_state_sync_after_write_set(state, opts->sync_after_write);
pv_state_direct_io_set(state, opts->direct_io);
pv_state_rate_limit_set(state, opts->rate_limit);
pv_state_target_buffer_size_set(state, opts->buffer_size);
pv_state_no_splice_set(state, opts->no_splice);
src/main/options.c
+5 -1
View File
@@ -83,6 +83,7 @@ opts_t opts_parse(int argc, char **argv)
{"skip-errors", 0, NULL, (int) 'E'},
{"stop-at-size", 0, NULL, (int) 'S'},
{"sync", 0, NULL, (int) 'Y'},
{"direct-io", 0, NULL, (int) 'K'},
{"remote", 1, NULL, (int) 'R'},
{"pidfile", 1, NULL, (int) 'P'},
{"watchfd", 1, NULL, (int) 'd'},
@@ -92,7 +93,7 @@ opts_t opts_parse(int argc, char **argv)
int option_index = 0;
#endif
char *short_options =
"hVpteIrab8TA:fnqcWD:s:l0i:w:H:N:F:L:B:CESYR:P:d:m:";
"hVpteIrab8TA:fnqcWD:s:l0i:w:H:N:F:L:B:CESYKR:P:d:m:";
int c, numopts;
unsigned int check_pid;
int check_fd;
@@ -337,6 +338,9 @@ opts_t opts_parse(int argc, char **argv)
case 'Y':
opts->sync_after_write = true;
break;
case 'K':
opts->direct_io = true;
break;
case 'R':
opts->remote = pv_getnum_ui(optarg);
break;
src/nls/pv.pot
+37 -33
View File
@@ -8,7 +8,7 @@ msgid ""
msgstr ""
"Project-Id-Version: PACKAGE VERSION\n"
"Report-Msgid-Bugs-To: \n"
"POT-Creation-Date: 2023-07-26 22:09+0100\n"
"POT-Creation-Date: 2023-07-27 20:49+0100\n"
"PO-Revision-Date: YEAR-MO-DA HO:MI+ZONE\n"
"Last-Translator: FULL NAME <EMAIL@ADDRESS>\n"
"Language-Team: LANGUAGE <LL@li.org>\n"
@@ -44,7 +44,7 @@ msgstr ""
msgid "failed to read file"
msgstr ""
#: src/pv/file.c:253 src/main/options.c:292
#: src/pv/file.c:253 src/main/options.c:293
msgid "failed to stat file"
msgstr ""
@@ -84,11 +84,11 @@ msgstr ""
msgid "write failed"
msgstr ""
#: src/pv/transfer.c:726 src/pv/display.c:646
#: src/pv/transfer.c:808 src/pv/display.c:646
msgid "buffer allocation failed"
msgstr ""
#: src/pv/transfer.c:817
#: src/pv/transfer.c:913
msgid "select call failed"
msgstr ""
@@ -96,7 +96,7 @@ msgstr ""
#: src/pv/watchpid.c:110 src/pv/watchpid.c:151 src/pv/watchpid.c:168
#: src/pv/watchpid.c:178 src/pv/watchpid.c:192 src/pv/watchpid.c:280
#: src/pv/watchpid.c:289 src/pv/watchpid.c:322 src/pv/watchpid.c:343
#: src/pv/loop.c:533 src/pv/loop.c:595 src/pv/loop.c:647
#: src/pv/loop.c:542 src/pv/loop.c:604 src/pv/loop.c:656
msgid "pid"
msgstr ""
@@ -331,55 +331,59 @@ msgid "flush cache to disk after every write"
msgstr ""
#: src/main/help.c:97
msgid "use direct I/O to bypass cache"
msgstr ""
#: src/main/help.c:99
msgid "PID"
msgstr ""
#: src/main/help.c:98
#: src/main/help.c:100
msgid "update settings of process PID"
msgstr ""
#: src/main/help.c:101
#: src/main/help.c:103
msgid "FILE"
msgstr ""
#: src/main/help.c:102
#: src/main/help.c:104
msgid "save process ID in FILE"
msgstr ""
#: src/main/help.c:104
#: src/main/help.c:106
msgid "PID[:FD]"
msgstr ""
#: src/main/help.c:105
#: src/main/help.c:107
msgid "watch file FD opened by process PID"
msgstr ""
#: src/main/help.c:108
#: src/main/help.c:110
msgid "show this help and exit"
msgstr ""
#: src/main/help.c:110
#: src/main/help.c:112
msgid "show version information and exit"
msgstr ""
#: src/main/help.c:117
#: src/main/help.c:119
#, c-format
msgid "Usage: %s [OPTION] [FILE]..."
msgstr ""
#: src/main/help.c:120
#: src/main/help.c:122
msgid ""
"Concatenate FILE(s), or standard input, to standard output,\n"
"with monitoring."
msgstr ""
#: src/main/help.c:213
#: src/main/help.c:215
msgid ""
"Debugging is enabled; export the DEBUG environment variable to define the\n"
"output filename.\n"
msgstr ""
#: src/main/help.c:217
#: src/main/help.c:219
#, c-format
msgid "Please report any bugs to %s."
msgstr ""
@@ -388,62 +392,62 @@ msgstr ""
msgid "state allocation failed"
msgstr ""
#: src/main/main.c:251
#: src/main/main.c:252
msgid "failed to read terminal attributes"
msgstr ""
#: src/main/options.c:105
#: src/main/options.c:106
msgid "option structure allocation failed"
msgstr ""
#: src/main/options.c:119
#: src/main/options.c:120
msgid "option structure argv allocation failed"
msgstr ""
#: src/main/options.c:164
#: src/main/options.c:165
msgid "integer argument expected"
msgstr ""
#: src/main/options.c:175
#: src/main/options.c:176
msgid "numeric argument expected"
msgstr ""
#: src/main/options.c:186
#: src/main/options.c:187
msgid "process ID or pid:fd pair expected"
msgstr ""
#: src/main/options.c:193
#: src/main/options.c:194
msgid "invalid process ID"
msgstr ""
#: src/main/options.c:362
#, c-format
msgid "Try `%s --help' for more information."
msgstr ""
#: src/main/options.c:366
#, c-format
msgid "Try `%s --help' for more information."
msgstr ""
#: src/main/options.c:370
#, c-format
msgid "Try `%s -h' for more information."
msgstr ""
#: src/main/options.c:382
#: src/main/options.c:386
msgid ""
"cannot use line mode or transfer modifier options when watching file "
"descriptors"
msgstr ""
#: src/main/options.c:390
#: src/main/options.c:394
msgid "cannot use cursor positioning when watching file descriptors"
msgstr ""
#: src/main/options.c:398
#: src/main/options.c:402
msgid "cannot use remote control when watching file descriptors"
msgstr ""
#: src/main/options.c:406
#: src/main/options.c:410
msgid "cannot transfer files when watching file descriptors"
msgstr ""
#: src/main/options.c:414
#: src/main/options.c:418
msgid "not available on systems without /proc/self/fdinfo"
msgstr ""
src/pv/file.c
+12
View File
@@ -293,6 +293,18 @@ int pv_next_file(pvstate_t state, int filenum, int oldfd)
if (0 == strcmp(state->input_files[filenum], "-")) {
state->current_file = "(stdin)";
}
#ifdef O_DIRECT
/*
* Set or clear O_DIRECT on the file descriptor.
*/
fcntl(fd, F_SETFL,
(state->direct_io ? O_DIRECT : 0) | fcntl(fd, F_GETFL));
/*
* We don't clear direct_io_changed here, to avoid race conditions
* that could cause the input and output settings to differ.
*/
#endif /* O_DIRECT */
return fd;
}
src/pv/loop.c
+9
View File
@@ -117,6 +117,15 @@ int pv_main_loop(pvstate_t state)
pv_crs_fini(state);
return state->exit_status;
}
#ifdef O_DIRECT
/*
* Set or clear O_DIRECT on the output.
*/
fcntl(STDOUT_FILENO, F_SETFL,
(state->direct_io ? O_DIRECT : 0) | fcntl(STDOUT_FILENO,
F_GETFL));
state->direct_io_changed = false;
#endif /* O_DIRECT */
/*
* Set target buffer size if the initial file's block size can be
src/pv/state.c
+6
View File
@@ -206,6 +206,12 @@ void pv_state_sync_after_write_set(pvstate_t state, bool val)
state->sync_after_write = val;
};
void pv_state_direct_io_set(pvstate_t state, bool val)
{
state->direct_io = val;
state->direct_io_changed = true;
};
void pv_state_rate_limit_set(pvstate_t state, unsigned long long val)
{
state->rate_limit = val;
src/pv/transfer.c
+101 -5
View File
@@ -684,6 +684,63 @@ static int pv__transfer_write(pvstate_t state,
}
/*
* Return a pointer to a newly allocated buffer of the given size, aligned
* appropriately for the current input and output file descriptors
* (important if using O_DIRECT).
*
* Falls back to unaligned allocation if it was not possible to get an
* aligned buffer, or if the relevant operating system features were not
* available. With O_DIRECT, this means that transfers could fail with an
* "Invalid argument" error (EINVAL).
*
* Returns NULL on complete allocation failure.
*/
static unsigned char *pv__allocate_aligned_buffer(int fd,
size_t target_size)
{
unsigned char *newptr;
#if defined(HAVE_FPATHCONF) && defined(HAVE_POSIX_MEMALIGN) && defined(_PC_REC_XFER_ALIGN)
long input_alignment, output_alignment, min_alignment;
size_t required_alignment;
input_alignment = fd >= 0 ? fpathconf(fd, _PC_REC_XFER_ALIGN) : -1;
output_alignment = fpathconf(STDOUT_FILENO, _PC_REC_XFER_ALIGN);
# if defined(HAVE_SYSCONF) && defined(_SC_PAGESIZE)
min_alignment = sysconf(_SC_PAGESIZE);
# else /* ! defined(HAVE_SYSCONF) && defined(_SC_PAGESIZE) */
min_alignment = 8192;
# endif /* defined(HAVE_SYSCONF) && defined(_SC_PAGESIZE) */
if (input_alignment > output_alignment) {
required_alignment = input_alignment;
} else if (output_alignment > input_alignment) {
required_alignment = output_alignment;
} else if (input_alignment < min_alignment) {
required_alignment = min_alignment;
} else {
required_alignment = input_alignment;
}
/* Ensure the alignment is at least the page size. */
if (required_alignment < min_alignment) {
required_alignment = min_alignment;
}
if (0 !=
posix_memalign((void **) (&newptr), required_alignment,
target_size)) {
newptr = (unsigned char *) malloc(target_size);
}
#else /* ! defined(HAVE_FPATHCONF) && defined(HAVE_POSIX_MEMALIGN) && defined(_PC_REC_XFER_ALIGN) */
newptr = (unsigned char *) malloc(target_size);
#endif /* defined(HAVE_FPATHCONF) && defined(HAVE_POSIX_MEMALIGN) && defined(_PC_REC_XFER_ALIGN) */
return newptr;
}
/*
* Transfer some data from "fd" to standard output, timing out after 9/100
* of a second. If state->rate_limit is >0, and/or "allowed" is >0, only up
@@ -707,6 +764,26 @@ long pv_transfer(pvstate_t state, int fd, int *eof_in, int *eof_out,
if (NULL == state)
return 0;
#ifdef O_DIRECT
/*
* Set or clear O_DIRECT on the input and output file descriptors,
* if the setting has changed.
*/
if (state->direct_io_changed) {
if (!(*eof_in)) {
fcntl(fd, F_SETFL,
(state->direct_io ? O_DIRECT : 0) | fcntl(fd,
F_GETFL));
}
if (!(*eof_out)) {
fcntl(STDOUT_FILENO, F_SETFL,
(state->direct_io ? O_DIRECT : 0) |
fcntl(STDOUT_FILENO, F_GETFL));
}
state->direct_io_changed = false;
}
#endif /* O_DIRECT */
/*
* Reinitialise the error skipping variables if the file descriptor
* has changed since the last time we were called.
@@ -717,10 +794,15 @@ long pv_transfer(pvstate_t state, int fd, int *eof_in, int *eof_out,
state->read_error_warning_shown = 0;
}
/*
* Allocate a new buffer, aligned appropriately for the input file
* (important if using O_DIRECT).
*/
if (NULL == state->transfer_buffer) {
state->buffer_size = state->target_buffer_size;
state->transfer_buffer =
(unsigned char *) malloc(state->buffer_size + 32);
pv__allocate_aligned_buffer(fd,
state->target_buffer_size +
32);
if (NULL == state->transfer_buffer) {
pv_error(state, "%s: %s",
_("buffer allocation failed"),
@@ -728,16 +810,22 @@ long pv_transfer(pvstate_t state, int fd, int *eof_in, int *eof_out,
state->exit_status |= 64;
return -1;
}
state->buffer_size = state->target_buffer_size;
}
/*
* Reallocate the buffer if the buffer size has changed mid-transfer.
* Reallocate the buffer if the buffer size has changed
* mid-transfer. We have to do this by allocating a new buffer,
* copying to it, and freeing the old one (potentially leaking
* memory) because the buffer may need to be aligned for O_DIRECT,
* and we can't realloc() an aligned buffer.
*/
if (state->buffer_size < state->target_buffer_size) {
unsigned char *newptr;
newptr =
realloc(state->transfer_buffer,
state->target_buffer_size + 32);
pv__allocate_aligned_buffer(fd,
state->target_buffer_size +
32);
if (NULL == newptr) {
/*
* Reset target if realloc failed so we don't keep
@@ -748,6 +836,14 @@ long pv_transfer(pvstate_t state, int fd, int *eof_in, int *eof_out,
} else {
debug("%s: %ld", "buffer resized",
state->buffer_size);
/*
* Copy the old buffer contents into the new buffer,
* and free the old one.
*/
if (state->buffer_size > 0)
memcpy(newptr, state->transfer_buffer,
state->buffer_size);
free(state->transfer_buffer);
state->transfer_buffer = newptr;
state->buffer_size = state->target_buffer_size;
}