[于 2018-01-13 重写。]
标准 I/O(printf() 等)在将数字数据转换为文本形式方面确实相对较慢。这里的问题是输出表格的行
其中三个都是十进制表示法的无符号(32 位)整数,或者 -1。为简单起见,我们将值 UINT32_MAX (4294967295) 保留为 -1。
我建议采用两种方法:
从右到左构造每条记录。这避免了检查数字中有多少位的需要。
-
一次缓冲多条记录。这减少了fwrite() 调用的数量,代价是适度的动态分配缓冲区。
请注意,这意味着每个块中的记录必须从后到先处理,以保持正确的顺序。
考虑以下代码。请注意,我已将node_t 和dijkstra_t 的定义缩减为实际使用的字段,以便可以按原样编译以下示例。另请注意,必须使用UINT32_MAX,而不是-1 for parent 或cost,因为它们的类型现在是uint32_t。
#include <stdlib.h>
#include <stdint.h>
#include <limits.h>
#include <stdio.h>
typedef struct {
uint32_t parent; /* Use UINT32_MAX for -1 */
uint32_t cost; /* Use UINT32_MAX for -1 */
} node_t;
typedef struct {
node_t *nodes;
uint32_t num_nodes;
} dijkstra_t;
/* This function will store an unsigned 32-bit value
in decimal form, ending at 'end'.
UINT32_MAX will be written as "-1", however.
Returns a pointer to the start of the value.
*/
static inline char *prepend_value(char *end, uint32_t value)
{
if (value == UINT32_MAX) {
*(--end) = '1';
*(--end) = '-';
} else {
do {
*(--end) = '0' + (value % 10u);
value /= 10u;
} while (value);
}
return end;
}
/* Each record consists of three unsigned 32-bit integers,
each at most 10 characters, with spaces in between
and a newline at end. Thus, at most 33 characters. */
#define RECORD_MAXLEN 33
/* We process records in chunks of 16384.
Maximum number of records (nodes) is 2**32 - 2 - RECORD_CHUNK,
or 4,294,950,910 in this case. */
#define RECORD_CHUNK 16384
/* Each chunk of record is up to CHUNK_CHARS long.
(Roughly half a megabyte in this case.) */
#define CHUNK_CHARS (RECORD_MAXLEN * RECORD_CHUNK)
/* Save the edges in a graph to a stream.
Returns 0 if success, -1 if an error occurs.
*/
int save_edges(dijkstra_t *dij, FILE *out)
{
if (dij && out && !ferror(out)) {
const int nodes = dij->num_nodes;
const node_t *node = dij->nodes;
const uint32_t root_parent = dij->nodes->parent;
const uint32_t root_cost = dij->nodes->cost;
char *buf, *end, *ptr;
uint32_t o;
/* Allocate memory for the chunk buffer. */
buf = malloc(CHUNK_CHARS);
if (!buf)
return -1;
end = buf + CHUNK_CHARS;
/* Temporarily, we reset the root node parent
to UINT32_MAX and cost to 0, so that the
very first record in the output is "0 0 -1". */
dij->nodes->cost = 0;
dij->nodes->parent = UINT32_MAX;
for (o = 0; o < nodes; o += RECORD_CHUNK) {
uint32_t i = (o + RECORD_CHUNK < nodes) ? o + RECORD_CHUNK : nodes;
/* Fill buffer back-to-front. */
ptr = end;
while (i-->o) {
const node_t *curr = node + i;
/* Format: <i> ' ' <cost> ' ' <parent> '\n' */
/* We construct the record from right to left. */
*(--ptr) = '\n';
ptr = prepend_value(ptr, curr->parent);
*(--ptr) = ' ';
ptr = prepend_value(ptr, curr->cost);
*(--ptr) = ' ';
ptr = prepend_value(ptr, i);
}
/* Write the chunk buffer out. */
if (fwrite(ptr, 1, (size_t)(end - ptr), out) != (size_t)(end - ptr)) {
dij->nodes->cost = root_cost;
dij->nodes->parent = root_parent;
free(buf);
return -1;
}
}
/* Reset root node, and free the buffer. */
dij->nodes->cost = root_cost;
dij->nodes->parent = root_parent;
free(buf);
/* Check for write errors. */
if (fflush(out))
return -1;
if (ferror(out))
return -1;
/* Success. */
return 0;
}
return -1;
}
如果我们可以使用 POSIX 低级 I/O(<unistd.h> 中的open()、close()、write() 和 fstat()),则可以实现额外的加速。当目的地是管道或设备时,我们可以直接写入数据;当目标是文件时,我们应该写入st_blksize 的倍数块,以避免读取-修改-写入循环。与标准 I/O 不同,对于低级 I/O,我们只需一个“溢出”缓冲区st_blksize 就可以做到这一点,而无需在内存中复制整个块缓冲区。但是,由于该问题未标记为posix,因此我将避免沿这些边缘进行进一步讨论。
OP 表示他们自己的版本仍然更快。我发现这很难相信,因为它比我上面的版本做得更多。当我检查时,在我的机器上,一个大型数据集(比如 100,000,000)不能在单个 fwrite() 调用中写入,因为它只进行部分写入;实际写入整个数据集需要一个循环。因此,在我看来,OP 用来比较不同版本的基准是非常可疑的。
请考虑以下微基准测试。它生成一个单链表,并使用外部编译的save_graph() 函数将其输出(到标准输出)。实现了三个版本:null,根本不保存任何东西; antonkretov,用于 OP 的实现(适用于此处工作);和 nominalanimal,我的。
生成文件:
CC := gcc
CFLAGS := -std=c99 -O2 -Wall
LDFLAGS :=
BINS := test-null test-antonkretov test-nominalanimal
NODES := 100000000
.PHONY: all clean run
all: clean $(BINS)
clean:
rm -f $(BINS) *.o
%.o: %.c
$(CC) $(CFLAGS) -c $^
test-null: main.o data-null.o
$(CC) $(CFLAGS) $^ -o $@
test-antonkretov: main.o data-antonkretov.o
$(CC) $(CFLAGS) $^ -o $@
test-nominalanimal: main.o data-nominalanimal.o
$(CC) $(CFLAGS) $^ -o $@
run: $(BINS)
@echo "Testing $(NODES) nodes."
@./test-null $(NODES) > /dev/null
@echo "Overhead (nothing saved):"
@bash -c 'time ./test-null $(NODES) > /dev/null'
@echo ""
@echo "Anton Kretov:"
@bash -c 'time ./test-antonkretov $(NODES) > /dev/null'
@echo ""
@echo "Nominal Animal:"
@bash -c 'time ./test-nominalanimal $(NODES) > /dev/null'
@echo ""
请注意,本论坛将Tabs转换为空格,而Makefile格式需要缩进才能使用空格,因此如果将上述内容复制并粘贴到文件中,则需要运行例如sed -e 's|^ *|\t|' -i Makefile 修复它。
data.h:
#ifndef DATA_H
#define DATA_H
#include <stdint.h>
#include <limits.h>
#include <stdio.h>
#define INVALID_COST UINT32_MAX
#define INVALID_PARENT UINT32_MAX
typedef struct {
uint32_t parent; /* Use INVALID_PARENT for -1 */
uint32_t cost; /* Use INVALID_COST for -1 */
} node_t;
typedef struct {
node_t *nodes;
uint32_t num_nodes;
} dijkstra_t;
int save_graph(dijkstra_t *, FILE *);
#endif /* DATA_H */
data-null.c,用于测量运行时开销:
#include "data.h"
int save_graph(dijkstra_t *dij, FILE *out)
{
/* Does not do anything */
return 0;
}
data-antonkretov.c,OP 的保存例程版本,用于比较:
#include <stdlib.h>
#include "data.h"
int getNumberOfDigits(uint32_t x)
{
if (x >= 10000) {
if (x >= 10000000) {
if (x >= 100000000) {
if (x >= 1000000000)
return 9;
return 8;
}
return 7;
}
if (x >= 100000) {
if (x >= 1000000)
return 6;
return 5;
}
return 4;
}
if (x >= 100) {
if (x >= 1000)
return 3;
return 2;
}
if (x >= 10)
return 1;
return 0;
}
int save_graph(dijkstra_t *dij, FILE *out)
{
uint32_t numberOfNodes = dij->num_nodes;
size_t bufferLength = numberOfNodes * (size_t)33;
size_t bufferCounter = 0, counter;
size_t bytes;
uint32_t number, digits, i;
char *buffer;
if ((size_t)(bufferLength / 33) != numberOfNodes)
return -1;
buffer = malloc(bufferLength);
if (!buffer)
return -1;
buffer[bufferCounter++] = '0';
buffer[bufferCounter++] = ' ';
buffer[bufferCounter++] = '0';
buffer[bufferCounter++] = ' ';
buffer[bufferCounter++] = '-';
buffer[bufferCounter++] = '1';
buffer[bufferCounter++] = '\n';
for (i = 1; i < numberOfNodes; i++) {
const node_t *const node = dij->nodes + i;
number = i;
digits = getNumberOfDigits(number);
counter = bufferCounter;
do {
buffer[counter + digits] = '0' + (number % 10u);
--digits;
++bufferCounter;
} while (number /= 10u);
buffer[bufferCounter++] = ' ';
number = node->cost;
if (number != UINT32_MAX) {
digits = getNumberOfDigits(number);
counter = bufferCounter;
do {
buffer[counter + digits] = '0' + (number % 10u);
--digits;
++bufferCounter;
} while (number /= 10u);
} else {
buffer[bufferCounter++] = '-';
buffer[bufferCounter++] = '1';
}
buffer[bufferCounter++] = ' ';
number = node->parent;
if (number != UINT32_MAX) {
digits = getNumberOfDigits(number);
counter = bufferCounter;
do {
buffer[counter + digits] = '0' + (number % 10u);
--digits;
++bufferCounter;
} while (number /= 10u);
} else {
buffer[bufferCounter++] = '-';
buffer[bufferCounter++] = '1';
}
buffer[bufferCounter++] = '\n';
}
counter = 0;
while (counter < bufferCounter) {
bytes = fwrite(buffer + counter, 1, bufferCounter - counter, out);
if (!bytes) {
free(buffer);
return -1;
}
counter += bytes;
}
free(buffer);
return 0;
}
data-nominalanimal.c,我的分块从后到前版本的保存例程:
#include <stdlib.h>
#include "data.h"
/* This function will store an unsigned 32-bit value
in decimal form, ending at 'end'.
UINT32_MAX will be written as "-1", however.
Returns a pointer to the start of the value.
*/
static inline char *prepend_value(char *end, uint32_t value)
{
if (value == UINT32_MAX) {
*(--end) = '1';
*(--end) = '-';
} else {
do {
*(--end) = '0' + (value % 10u);
value /= 10u;
} while (value);
}
return end;
}
/* Each record consists of three unsigned 32-bit integers,
each at most 10 characters, with spaces in between
and a newline at end. Thus, at most 33 characters. */
#define RECORD_MAXLEN 33
/* We process records in chunks of 16384.
Maximum number of records (nodes) is 2**32 - 2 - RECORD_CHUNK,
or 4,294,950,910 in this case. */
#define RECORD_CHUNK 16384
/* Each chunk of record is up to CHUNK_CHARS long.
(Roughly half a megabyte in this case.) */
#define CHUNK_CHARS (RECORD_MAXLEN * RECORD_CHUNK)
/* Save the edges in a graph to a stream.
Returns 0 if success, -1 if an error occurs.
*/
int save_graph(dijkstra_t *dij, FILE *out)
{
if (dij && out && !ferror(out)) {
const int nodes = dij->num_nodes;
const node_t *node = dij->nodes;
const uint32_t root_parent = dij->nodes->parent;
const uint32_t root_cost = dij->nodes->cost;
char *buf, *end, *ptr;
uint32_t o;
/* Allocate memory for the chunk buffer. */
buf = malloc(CHUNK_CHARS);
if (!buf)
return -1;
end = buf + CHUNK_CHARS;
/* Temporarily, we reset the root node parent
to UINT32_MAX and cost to 0, so that the
very first record in the output is "0 0 -1". */
dij->nodes->cost = 0;
dij->nodes->parent = UINT32_MAX;
for (o = 0; o < nodes; o += RECORD_CHUNK) {
uint32_t i = (o + RECORD_CHUNK < nodes) ? o + RECORD_CHUNK : nodes;
/* Fill buffer back-to-front. */
ptr = end;
while (i-->o) {
const node_t *curr = node + i;
/* Format: <i> ' ' <cost> ' ' <parent> '\n' */
/* We construct the record from right to left. */
*(--ptr) = '\n';
ptr = prepend_value(ptr, curr->parent);
*(--ptr) = ' ';
ptr = prepend_value(ptr, curr->cost);
*(--ptr) = ' ';
ptr = prepend_value(ptr, i);
}
/* Write buffer. */
if (fwrite(ptr, 1, (size_t)(end - ptr), out) != (size_t)(end - ptr)) {
dij->nodes->cost = root_cost;
dij->nodes->parent = root_parent;
free(buf);
return -1;
}
}
/* Reset root node, and free the buffer. */
dij->nodes->cost = root_cost;
dij->nodes->parent = root_parent;
free(buf);
if (fflush(out))
return -1;
if (ferror(out))
return -1;
return 0;
}
return -1;
}
最后是主程序本身,main.c,它生成数据并调用save_graph() 函数:
#include <stdlib.h>
#include <inttypes.h>
#include <limits.h>
#include <string.h>
#include "data.h"
#define EDGES_MAX 4294901759
int main(int argc, char *argv[])
{
dijkstra_t graph;
size_t bytes;
uint32_t edges, i;
char dummy;
if (argc != 2 || !strcmp(argv[1], "-h") || !strcmp(argv[1], "--help")) {
fprintf(stderr, "\nUsage: %s EDGES\n\n", argv[0]);
return EXIT_SUCCESS;
}
if (sscanf(argv[1], " %" SCNu32 " %c", &edges, &dummy) != 1 || edges < 1 || edges > EDGES_MAX) {
fprintf(stderr, "%s: Invalid number of edges.\n", argv[1]);
return EXIT_FAILURE;
}
bytes = (1 + (size_t)edges) * sizeof graph.nodes[0];
if ((size_t)(bytes / (1 + (size_t)edges)) != sizeof graph.nodes[0]) {
fprintf(stderr, "%s: Too many edges.\n", argv[1]);
return EXIT_FAILURE;
}
graph.num_nodes = edges + 1;
graph.nodes = malloc(bytes);
if (!graph.nodes) {
fprintf(stderr, "%s: Too many edges: out of memory.\n", argv[1]);
return EXIT_FAILURE;
}
/* Generate a graph; no randomness, to keep timing steady. */
graph.nodes[0].parent = INVALID_COST;
graph.nodes[0].cost = 0;
for (i = 1; i <= edges; i++) {
graph.nodes[i].parent = i - 1;
graph.nodes[i].cost = 1 + (i % 10);
}
/* Print graph. */
if (save_graph(&graph, stdout)) {
fprintf(stderr, "Write error!\n");
return EXIT_FAILURE;
}
/* Done. */
return EXIT_SUCCESS;
}
运行make clean run(或make NODES=100000000 clean run)会重新编译基准测试并测量它们的运行时间,以获取具有 100,000,000 个节点的图。在我的机器上,输出是
Testing 100000000 nodes.
Overhead (nothing saved):
real0m0.514s
user0m0.297s
sys0m0.217s
Anton Kretov:
real0m4.059s
user0m3.379s
sys0m0.680s
Nominal Animal:
real0m3.336s
user0m3.151s
sys0m0.184s
这表明我的速度明显更快。如果我们忽略开销(生成图表),我的需要大约 2.8 秒的实时时间来将数据保存到/dev/null,而 OP 大约需要 3.5 秒。换句话说,我的速度提高了 20%。
请务必注意,这两个测试确实会产生完全相同的输出。例如,./test-nominalanimal 100000000 | sha256sum - 和 ./test-antonkretov 100000000 | sha256sum - 都显示完全相同的 SHA256 校验和,7504a1c97167701297c03c4aab8b0f20c5cac82a50128074d6e09c474353d0f8。
(您也可以将输出保存到文件中,然后比较它们;两者的长度正好是 1,987,777,795 字节,并且包含完全相同的数据。我确实检查了。)
如果您想运行将数据存储到存储器的基准测试,为了公平比较,您需要从冷缓存开始。否则,您运行基准测试的顺序将严重影响它们的时间安排。