TheCloud/f3
1
0
Fork 0
mirror of https://github.com/AltraMayor/f3.git synced 2025-09-18 11:37:59 -04:00
Code Issues Packages Projects Releases Wiki Activity

libflow: rename fields and variables for generality

Browse Source 
The original libflow was written for f3write only, and
this was reflected in the name of fields and variables
throughout the code. This patch renames these fields and
variables to reflect the fact that libflow will eventually
support f3write AND f3read.
This commit is contained in:
Michel Machado 2020-06-19 13:58:48 -04:00
parent a759140bc5
commit 79cd2d66a2
3 changed files with 56 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
f3write.c
View File

@ -150,13 +150,13 @@ static int write_all(int fd, const char *buf, size_t count)
return 0; return 0;
} }
static int write_chunk(int fd, ssize_t chunk_size, uint64_t *poffset) static int write_chunk(int fd, size_t chunk_size, uint64_t *poffset)
{ {
char buf[MAX_WRITE_SIZE]; char buf[MAX_BUFFER_SIZE];
while (chunk_size > 0) { while (chunk_size > 0) {
ssize_t turn_size = chunk_size <= MAX_WRITE_SIZE size_t turn_size = chunk_size <= MAX_BUFFER_SIZE
? chunk_size : MAX_WRITE_SIZE; ? chunk_size : MAX_BUFFER_SIZE;
int ret; int ret;
chunk_size -= turn_size; chunk_size -= turn_size;
*poffset = fill_buffer(buf, turn_size, *poffset); *poffset = fill_buffer(buf, turn_size, *poffset);
@ -203,10 +203,8 @@ static int create_and_fill_file(const char *path, long number, size_t size,
remaining = size; remaining = size;
start_measurement(fw); start_measurement(fw);
while (remaining > 0) { while (remaining > 0) {
ssize_t write_size = fw->block_size * size_t write_size = get_rem_chunk_size(fw);
(fw->blocks_per_delay - fw->written_blocks); if (write_size > remaining)
assert(write_size > 0);
if ((size_t)write_size > remaining)
write_size = remaining; write_size = remaining;
saved_errno = write_chunk(fd, write_size, &offset); saved_errno = write_chunk(fd, write_size, &offset);
if (saved_errno) if (saved_errno)

54
libflow.c
View File

@ -19,22 +19,22 @@ static inline void move_to_inc_at_start(struct flow *fw)
} }
void init_flow(struct flow *fw, uint64_t total_size, void init_flow(struct flow *fw, uint64_t total_size,
long max_write_rate, int progress, long max_process_rate, int progress,
flow_func_flush_chunk_t func_flush_chunk) flow_func_flush_chunk_t func_flush_chunk)
{ {
fw->total_size = total_size; fw->total_size = total_size;
fw->total_written = 0; fw->total_processed = 0;
fw->progress = progress; fw->progress = progress;
fw->block_size = 512; /* Bytes */ fw->block_size = 512; /* Bytes */
fw->blocks_per_delay = 1; /* 512B/s */ fw->blocks_per_delay = 1; /* 512B/s */
fw->delay_ms = 1000; /* 1s */ fw->delay_ms = 1000; /* 1s */
fw->max_write_rate = max_write_rate <= 0 fw->max_process_rate = max_process_rate <= 0
? DBL_MAX : max_write_rate * 1024.; ? DBL_MAX : max_process_rate * 1024.;
fw->measured_blocks = 0; fw->measured_blocks = 0;
fw->measured_time_ms = 0; fw->measured_time_ms = 0;
fw->erase = 0; fw->erase = 0;
fw->func_flush_chunk = func_flush_chunk; fw->func_flush_chunk = func_flush_chunk;
fw->written_blocks = 0; fw->processed_blocks = 0;
fw->acc_delay_us = 0; fw->acc_delay_us = 0;
assert(fw->block_size > 0); assert(fw->block_size > 0);
assert(fw->block_size % SECTOR_SIZE == 0); assert(fw->block_size % SECTOR_SIZE == 0);
@ -104,16 +104,16 @@ static void report_progress(struct flow *fw, double inst_speed)
* the initial free space isn't exactly reported * the initial free space isn't exactly reported
* by the kernel; this issue has been seen on Macs. * by the kernel; this issue has been seen on Macs.
*/ */
if (fw->total_size < fw->total_written) if (fw->total_size < fw->total_processed)
fw->total_size = fw->total_written; fw->total_size = fw->total_processed;
percent = (double)fw->total_written * 100 / fw->total_size; percent = (double)fw->total_processed * 100 / fw->total_size;
erase(fw->erase); erase(fw->erase);
fw->erase = printf("%.2f%% -- %.2f %s/s", fw->erase = printf("%.2f%% -- %.2f %s/s",
percent, inst_speed, unit); percent, inst_speed, unit);
assert(fw->erase > 0); assert(fw->erase > 0);
if (has_enough_measurements(fw)) if (has_enough_measurements(fw))
fw->erase += pr_time( fw->erase += pr_time(
(fw->total_size - fw->total_written) / (fw->total_size - fw->total_processed) /
get_avg_speed(fw)); get_avg_speed(fw));
fflush(stdout); fflush(stdout);
} }
@ -188,14 +188,14 @@ static inline int is_rate_above(const struct flow *fw,
long delay, double inst_speed) long delay, double inst_speed)
{ {
/* We use logical or here to enforce the lowest limit. */ /* We use logical or here to enforce the lowest limit. */
return delay > fw->delay_ms || inst_speed > fw->max_write_rate; return delay > fw->delay_ms || inst_speed > fw->max_process_rate;
} }
static inline int is_rate_below(const struct flow *fw, static inline int is_rate_below(const struct flow *fw,
long delay, double inst_speed) long delay, double inst_speed)
{ {
/* We use logical and here to enforce both limist. */ /* We use logical and here to enforce both limist. */
return delay <= fw->delay_ms && inst_speed < fw->max_write_rate; return delay <= fw->delay_ms && inst_speed < fw->max_process_rate;
} }
static inline int flush_chunk(const struct flow *fw, int fd) static inline int flush_chunk(const struct flow *fw, int fd)
@ -244,20 +244,20 @@ static inline uint64_t diff_timeval_us(const struct timeval *t1,
t2->tv_usec - t1->tv_usec; t2->tv_usec - t1->tv_usec;
} }
int measure(int fd, struct flow *fw, ssize_t written) int measure(int fd, struct flow *fw, ssize_t processed)
{ {
ldiv_t result = ldiv(written, fw->block_size); ldiv_t result = ldiv(processed, fw->block_size);
struct timeval t2; struct timeval t2;
int64_t delay; int64_t delay;
double bytes_k, inst_speed; double bytes_k, inst_speed;
assert(result.rem == 0); assert(result.rem == 0);
fw->written_blocks += result.quot; fw->processed_blocks += result.quot;
fw->total_written += written; fw->total_processed += processed;
if (fw->written_blocks < fw->blocks_per_delay) if (fw->processed_blocks < fw->blocks_per_delay)
return 0; return 0;
assert(fw->written_blocks == fw->blocks_per_delay); assert(fw->processed_blocks == fw->blocks_per_delay);
if (flush_chunk(fw, fd) < 0) if (flush_chunk(fw, fd) < 0)
return -1; /* Caller can read errno(3). */ return -1; /* Caller can read errno(3). */
@ -269,10 +269,12 @@ int measure(int fd, struct flow *fw, ssize_t written)
bytes_k = fw->blocks_per_delay * fw->block_size * 1000.0; bytes_k = fw->blocks_per_delay * fw->block_size * 1000.0;
inst_speed = bytes_k / delay; inst_speed = bytes_k / delay;
if (delay < fw->delay_ms && inst_speed > fw->max_write_rate) { if (delay < fw->delay_ms && inst_speed > fw->max_process_rate) {
/* Wait until inst_speed == fw->max_write_rate (if possible). */ /* Wait until inst_speed == fw->max_process_rate
double wait_ms = round((bytes_k - delay * fw->max_write_rate) * (if possible).
/ fw->max_write_rate); */
double wait_ms = round((bytes_k - delay * fw->max_process_rate)
/ fw->max_process_rate);
if (wait_ms < 0) { if (wait_ms < 0) {
/* Wait what is possible. */ /* Wait what is possible. */
@ -297,7 +299,7 @@ int measure(int fd, struct flow *fw, ssize_t written)
} }
/* Update mean. */ /* Update mean. */
fw->measured_blocks += fw->written_blocks; fw->measured_blocks += fw->processed_blocks;
fw->measured_time_ms += delay; fw->measured_time_ms += delay;
switch (fw->state) { switch (fw->state) {
@ -340,9 +342,9 @@ int measure(int fd, struct flow *fw, ssize_t written)
case FW_STEADY: { case FW_STEADY: {
if (delay <= fw->delay_ms) { if (delay <= fw->delay_ms) {
if (inst_speed < fw->max_write_rate) { if (inst_speed < fw->max_process_rate) {
move_to_inc(fw); move_to_inc(fw);
} else if (inst_speed > fw->max_write_rate) { } else if (inst_speed > fw->max_process_rate) {
move_to_dec(fw); move_to_dec(fw);
} }
} else if (fw->blocks_per_delay > 1) { } else if (fw->blocks_per_delay > 1) {
@ -359,7 +361,7 @@ int measure(int fd, struct flow *fw, ssize_t written)
report_progress(fw, inst_speed); report_progress(fw, inst_speed);
/* Reset accumulators. */ /* Reset accumulators. */
fw->written_blocks = 0; fw->processed_blocks = 0;
fw->acc_delay_us = 0; fw->acc_delay_us = 0;
__start_measurement(fw); __start_measurement(fw);
return 0; return 0;
@ -371,7 +373,7 @@ int end_measurement(int fd, struct flow *fw)
int saved_errno; int saved_errno;
int ret = 0; int ret = 0;
if (fw->written_blocks <= 0) if (fw->processed_blocks <= 0)
goto out; goto out;
if (flush_chunk(fw, fd) < 0) { if (flush_chunk(fw, fd) < 0) {

36
libflow.h
View File

@ -8,10 +8,10 @@ struct flow;
typedef int (*flow_func_flush_chunk_t)(const struct flow *fw, int fd); typedef int (*flow_func_flush_chunk_t)(const struct flow *fw, int fd);
struct flow { struct flow {
/* Total number of bytes to be written. */ /* Total number of bytes to be processed. */
uint64_t total_size; uint64_t total_size;
/* Total number of bytes already written. */ /* Total number of bytes already processed. */
uint64_t total_written; uint64_t total_processed;
/* If true, show progress. */ /* If true, show progress. */
int progress; int progress;
/* Block size in bytes. */ /* Block size in bytes. */
@ -20,10 +20,10 @@ struct flow {
unsigned int delay_ms; unsigned int delay_ms;
/* Increment to apply to @blocks_per_delay. */ /* Increment to apply to @blocks_per_delay. */
int64_t step; int64_t step;
/* Blocks to write before measurement. */ /* Blocks to process before measurement. */
int64_t blocks_per_delay; int64_t blocks_per_delay;
/* Maximum write rate in bytes per second. */ /* Maximum processing rate in bytes per second. */
double max_write_rate; double max_process_rate;
/* Number of measured blocks. */ /* Number of measured blocks. */
uint64_t measured_blocks; uint64_t measured_blocks;
/* Measured time. */ /* Measured time. */
@ -42,10 +42,10 @@ struct flow {
* Initialized while measuring * Initialized while measuring
*/ */
/* Number of blocks written since last measurement. */ /* Number of blocks processed since last measurement. */
int64_t written_blocks; int64_t processed_blocks;
/* /*
* Accumulated delay before @written_blocks reaches @blocks_per_delay * Accumulated delay before @processed_blocks reaches @blocks_per_delay
* in microseconds. * in microseconds.
*/ */
uint64_t acc_delay_us; uint64_t acc_delay_us;
@ -55,15 +55,15 @@ struct flow {
struct timeval t1; struct timeval t1;
}; };
/* If @max_write_rate <= 0, the maximum write rate is infinity. /* If @max_process_rate <= 0, the maximum processing rate is infinity.
* The unit of @max_write_rate is KB per second. * The unit of @max_process_rate is KB per second.
*/ */
void init_flow(struct flow *fw, uint64_t total_size, void init_flow(struct flow *fw, uint64_t total_size,
long max_write_rate, int progress, long max_process_rate, int progress,
flow_func_flush_chunk_t func_flush_chunk); flow_func_flush_chunk_t func_flush_chunk);
void start_measurement(struct flow *fw); void start_measurement(struct flow *fw);
int measure(int fd, struct flow *fw, ssize_t written); int measure(int fd, struct flow *fw, ssize_t processed);
int end_measurement(int fd, struct flow *fw); int end_measurement(int fd, struct flow *fw);
static inline int has_enough_measurements(const struct flow *fw) static inline int has_enough_measurements(const struct flow *fw)
@ -84,6 +84,14 @@ static inline double get_avg_speed(struct flow *fw)
return get_avg_speed_given_time(fw, fw->measured_time_ms); return get_avg_speed_given_time(fw, fw->measured_time_ms);
} }
#define MAX_WRITE_SIZE (1<<21) /* 2MB */ static inline size_t get_rem_chunk_size(struct flow *fw)
{
ssize_t ret = (fw->blocks_per_delay - fw->processed_blocks) *
fw->block_size;
assert(ret > 0);
return ret;
}
#define MAX_BUFFER_SIZE (1<<21) /* 2MB */
#endif /* HEADER_LIBFLOW_H */ #endif /* HEADER_LIBFLOW_H */