pdxx
/
libserverframe
mirror of https://github.com/happyfish100/libserverframe.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity
libserverframe/src/sf_file_writer.c

708 lines
21 KiB
C
Raw Blame History

/*
* Copyright (c) 2020 YuQing <384681@qq.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <limits.h>
#include <fcntl.h>
#include <pthread.h>
#include "fastcommon/logger.h"
#include "fastcommon/sockopt.h"
#include "fastcommon/shared_func.h"
#include "fastcommon/pthread_func.h"
#include "fastcommon/sched_thread.h"
#include "sf_global.h"
#include "sf_func.h"
#include "sf_file_writer.h"
#define BINLOG_INDEX_ITEM_START_INDEX_STR "start_index"
#define BINLOG_INDEX_ITEM_START_INDEX_LEN \
(sizeof(BINLOG_INDEX_ITEM_START_INDEX_STR) - 1)
#define BINLOG_INDEX_ITEM_CURRENT_WRITE_STR "current_write"
#define BINLOG_INDEX_ITEM_CURRENT_WRITE_LEN \
(sizeof(BINLOG_INDEX_ITEM_CURRENT_WRITE_STR) - 1)
#define BINLOG_INDEX_ITEM_CURRENT_COMPRESS_STR "current_compress"
#define BINLOG_INDEX_ITEM_CURRENT_COMPRESS_LEN \
(sizeof(BINLOG_INDEX_ITEM_CURRENT_COMPRESS_STR) - 1)
static inline void sf_file_writer_get_binlog_filename(SFFileWriterInfo *writer)
{
sf_file_writer_get_filename_ex(
writer->cfg.data_path, writer->cfg.subdir_name,
writer->cfg.file_prefix, writer->binlog.last_index,
writer->file.name.str, writer->file.name.size);
}
static inline void sf_file_writer_get_index_filename_ex(
const char *data_path, const char *subdir_name,
const char *file_prefix, const int file_prefix_len,
char *filename, const int size)
{
#define INDEX_FILENAME_AFFIX_STR "_index.dat"
#define INDEX_FILENAME_AFFIX_LEN (sizeof(INDEX_FILENAME_AFFIX_STR) - 1)
char *p;
int data_path_len;
int subdir_name_len;
data_path_len = strlen(data_path);
subdir_name_len = strlen(subdir_name);
if (data_path_len + 1 + subdir_name_len + 1 + file_prefix_len +
INDEX_FILENAME_AFFIX_LEN >= size)
{
*filename = '\0';
return;
}
memcpy(filename, data_path, data_path_len);
p = filename + data_path_len;
*p++ = '/';
memcpy(p, subdir_name, subdir_name_len);
p += subdir_name_len;
*p++ = '/';
memcpy(p, file_prefix, file_prefix_len);
p += file_prefix_len;
memcpy(p, INDEX_FILENAME_AFFIX_STR, INDEX_FILENAME_AFFIX_LEN);
p += INDEX_FILENAME_AFFIX_LEN;
*p = '\0';
}
const char *sf_file_writer_get_index_filename(const char *data_path,
const char *subdir_name, char *filename, const int size)
{
sf_file_writer_get_index_filename_ex(data_path, subdir_name,
SF_BINLOG_FILE_PREFIX_STR, SF_BINLOG_FILE_PREFIX_LEN,
filename, size);
return filename;
}
int sf_file_writer_write_to_binlog_index_file_ex(const char *data_path,
const char *subdir_name, const char *file_prefix,
const int start_index, const int last_index,
const int compress_index)
{
char filename[PATH_MAX];
char buff[256];
char *p;
int result;
int len;
sf_file_writer_get_index_filename_ex(data_path, subdir_name, file_prefix,
strlen(file_prefix), filename, sizeof(filename));
p = buff;
memcpy(p, BINLOG_INDEX_ITEM_START_INDEX_STR,
BINLOG_INDEX_ITEM_START_INDEX_LEN);
p += BINLOG_INDEX_ITEM_START_INDEX_LEN;
*p++ = '=';
p += fc_itoa(start_index, p);
*p++ = '\n';
memcpy(p, BINLOG_INDEX_ITEM_CURRENT_WRITE_STR,
BINLOG_INDEX_ITEM_CURRENT_WRITE_LEN);
p += BINLOG_INDEX_ITEM_CURRENT_WRITE_LEN;
*p++ = '=';
p += fc_itoa(last_index, p);
*p++ = '\n';
memcpy(p, BINLOG_INDEX_ITEM_CURRENT_COMPRESS_STR,
BINLOG_INDEX_ITEM_CURRENT_COMPRESS_LEN);
p += BINLOG_INDEX_ITEM_CURRENT_COMPRESS_LEN;
*p++ = '=';
p += fc_itoa(compress_index, p);
*p++ = '\n';
len = p - buff;
if ((result=safeWriteToFile(filename, buff, len)) != 0) {
logError("file: "__FILE__", line: %d, "
"write to file \"%s\" fail, errno: %d, error info: %s",
__LINE__, filename, result, STRERROR(result));
}
return result;
}
static inline int write_to_binlog_index_file(SFFileWriterInfo *writer)
{
return sf_file_writer_write_to_binlog_index_file_ex(
writer->cfg.data_path, writer->cfg.subdir_name,
writer->cfg.file_prefix, writer->binlog.start_index,
writer->binlog.last_index, writer->binlog.compress_index);
}
static int get_binlog_info_from_file(const char *data_path,
const char *subdir_name, int *start_index,
int *last_index, int *compress_index)
{
char full_filename[PATH_MAX];
IniContext ini_context;
int result;
sf_file_writer_get_index_filename_ex(data_path, subdir_name,
SF_BINLOG_FILE_PREFIX_STR, SF_BINLOG_FILE_PREFIX_LEN,
full_filename, sizeof(full_filename));
if (access(full_filename, F_OK) != 0) {
return errno != 0 ? errno : EPERM;
}
if ((result=iniLoadFromFile(full_filename, &ini_context)) != 0) {
logError("file: "__FILE__", line: %d, "
"load from file \"%s\" fail, error code: %d",
__LINE__, full_filename, result);
return result;
}
*start_index = iniGetIntValue(NULL,
BINLOG_INDEX_ITEM_START_INDEX_STR,
&ini_context, 0);
*last_index = iniGetIntValue(NULL,
BINLOG_INDEX_ITEM_CURRENT_WRITE_STR,
&ini_context, 0);
*compress_index = iniGetIntValue(NULL,
BINLOG_INDEX_ITEM_CURRENT_COMPRESS_STR,
&ini_context, 0);
iniFreeContext(&ini_context);
return 0;
}
int sf_file_writer_get_binlog_indexes(const char *data_path,
const char *subdir_name, int *start_index, int *last_index)
{
int result;
int compress_index;
result = get_binlog_info_from_file(data_path, subdir_name,
start_index, last_index, &compress_index);
if (result == ENOENT) {
*start_index = *last_index = 0;
return 0;
} else {
return result;
}
}
static inline int get_binlog_index_from_file(SFFileWriterInfo *writer)
{
int result;
result = get_binlog_info_from_file(writer->cfg.data_path,
writer->cfg.subdir_name, &writer->binlog.start_index,
&writer->binlog.last_index, &writer->binlog.compress_index);
if (result == ENOENT) {
writer->binlog.start_index = 0;
writer->binlog.last_index = 0;
writer->binlog.compress_index = 0;
if (writer->cfg.file_rotate_size > 0) {
return write_to_binlog_index_file(writer);
} else {
return 0;
}
} else {
return result;
}
}
static int open_writable_binlog(SFFileWriterInfo *writer)
{
if (writer->file.fd >= 0) {
close(writer->file.fd);
}
sf_file_writer_get_binlog_filename(writer);
writer->file.fd = open(writer->file.name.str, O_WRONLY |
O_CREAT | O_APPEND | O_CLOEXEC, 0644);
if (writer->file.fd < 0) {
logError("file: "__FILE__", line: %d, "
"open file \"%s\" fail, "
"errno: %d, error info: %s",
__LINE__, writer->file.name.str,
errno, STRERROR(errno));
return errno != 0 ? errno : EACCES;
}
writer->file.size = lseek(writer->file.fd, 0, SEEK_END);
if (writer->file.size < 0) {
logError("file: "__FILE__", line: %d, "
"lseek file \"%s\" fail, "
"errno: %d, error info: %s",
__LINE__, writer->file.name.str,
errno, STRERROR(errno));
return errno != 0 ? errno : EIO;
}
return 0;
}
static int open_next_binlog(SFFileWriterInfo *writer)
{
sf_file_writer_get_binlog_filename(writer);
if (access(writer->file.name.str, F_OK) == 0) {
char bak_filename[PATH_MAX];
char date_str[32];
formatDatetime(g_current_time, "%Y%m%d%H%M%S",
date_str, sizeof(date_str));
fc_combine_two_strings(writer->file.name.str,
date_str, '.', bak_filename);
if (rename(writer->file.name.str, bak_filename) == 0) {
logWarning("file: "__FILE__", line: %d, "
"binlog file %s exist, rename to %s",
__LINE__, writer->file.name.str, bak_filename);
} else {
logError("file: "__FILE__", line: %d, "
"rename binlog %s to backup %s fail, "
"errno: %d, error info: %s",
__LINE__, writer->file.name.str, bak_filename,
errno, STRERROR(errno));
return errno != 0 ? errno : EPERM;
}
}
return open_writable_binlog(writer);
}
static int do_write_to_file(SFFileWriterInfo *writer,
char *buff, const int len)
{
int result;
if (fc_safe_write(writer->file.fd, buff, len) != len) {
result = errno != 0 ? errno : EIO;
logError("file: "__FILE__", line: %d, "
"write to binlog file \"%s\" fail, "
"errno: %d, error info: %s",
__LINE__, writer->file.name.str,
result, STRERROR(result));
return result;
}
if (writer->cfg.call_fsync) {
if (fsync(writer->file.fd) != 0) {
result = errno != 0 ? errno : EIO;
logError("file: "__FILE__", line: %d, "
"fsync to binlog file \"%s\" fail, errno: %d, "
"error info: %s", __LINE__, writer->file.name.str,
result, STRERROR(result));
return result;
}
}
writer->file.size += len;
if (writer->write_done_callback.func != NULL) {
writer->write_done_callback.func(writer,
writer->write_done_callback.args);
}
return 0;
}
int sf_file_writer_direct_write(SFFileWriterInfo *writer,
char *buff, const int len)
{
int result;
if ((writer->cfg.file_rotate_size <= 0) || (writer->file.size
+ len <= writer->cfg.file_rotate_size))
{
return do_write_to_file(writer, buff, len);
}
writer->binlog.last_index++; //binlog rotate
if ((result=write_to_binlog_index_file(writer)) == 0) {
result = open_next_binlog(writer);
}
if (result != 0) {
logError("file: "__FILE__", line: %d, "
"open binlog file \"%s\" fail",
__LINE__, writer->file.name.str);
return result;
}
return do_write_to_file(writer, buff, len);
}
int sf_file_writer_flush(SFFileWriterInfo *writer)
{
int result;
int len;
len = SF_BINLOG_BUFFER_PRODUCER_DATA_LENGTH(writer->binlog_buffer);
if (len == 0) {
return 0;
}
if ((result=sf_file_writer_direct_write(writer, writer->
binlog_buffer.buff, len)) == 0)
{
if (writer->flags & SF_FILE_WRITER_FLAGS_WANT_DONE_VERSION) {
writer->last_versions.done = writer->last_versions.pending;
}
}
writer->binlog_buffer.data_end = writer->binlog_buffer.buff;
return result;
}
int sf_file_writer_fsync(SFFileWriterInfo *writer)
{
int result;
if ((result=sf_file_writer_flush(writer)) != 0) {
return result;
}
if (fsync(writer->file.fd) == 0) {
return 0;
} else {
result = errno != 0 ? errno : EIO;
logError("file: "__FILE__", line: %d, "
"fsync to binlog file \"%s\" fail, errno: %d, "
"error info: %s", __LINE__, writer->file.name.str,
result, STRERROR(result));
return result;
}
}
int sf_file_writer_get_indexes(SFFileWriterInfo *writer,
int *start_index, int *last_index)
{
int result;
if (writer == NULL) { //for data recovery
*start_index = *last_index = 0;
return 0;
}
if (writer->binlog.last_index < 0) {
if ((result=get_binlog_index_from_file(writer)) != 0) {
*start_index = *last_index = -1;
return result;
}
}
*start_index = writer->binlog.start_index;
*last_index = writer->binlog.last_index;
return 0;
}
int sf_file_writer_deal_versioned_buffer(SFFileWriterInfo *writer,
BufferInfo *buffer, const int64_t version)
{
int result;
if (buffer->length >= writer->binlog_buffer.size / 4) {
if (SF_BINLOG_BUFFER_PRODUCER_DATA_LENGTH(writer->binlog_buffer) > 0) {
if ((result=sf_file_writer_flush(writer)) != 0) {
return result;
}
}
if ((result=sf_file_writer_direct_write(writer, buffer->
buff, buffer->length)) == 0)
{
if (writer->flags & SF_FILE_WRITER_FLAGS_WANT_DONE_VERSION) {
writer->last_versions.pending = version;
writer->last_versions.done = version;
}
}
return result;
}
if (writer->cfg.file_rotate_size > 0 && writer->file.size +
SF_BINLOG_BUFFER_PRODUCER_DATA_LENGTH(writer->binlog_buffer) +
buffer->length > writer->cfg.file_rotate_size)
{
if ((result=sf_file_writer_flush(writer)) != 0) {
return result;
}
} else if (SF_BINLOG_BUFFER_PRODUCER_BUFF_REMAIN(
writer->binlog_buffer) < buffer->length)
{
if ((result=sf_file_writer_flush(writer)) != 0) {
return result;
}
}
if (writer->flags & SF_FILE_WRITER_FLAGS_WANT_DONE_VERSION) {
writer->last_versions.pending = version;
}
memcpy(writer->binlog_buffer.data_end, buffer->buff, buffer->length);
writer->binlog_buffer.data_end += buffer->length;
return 0;
}
int sf_file_writer_save_buffer_ex(SFFileWriterInfo *writer,
const int length, const bool flush)
{
int result;
if (writer->cfg.file_rotate_size > 0 && writer->file.size +
SF_BINLOG_BUFFER_PRODUCER_DATA_LENGTH(writer->binlog_buffer) +
length > writer->cfg.file_rotate_size)
{
if ((result=sf_file_writer_flush(writer)) != 0) {
return result;
}
}
writer->binlog_buffer.data_end += length;
if (flush || SF_BINLOG_BUFFER_PRODUCER_BUFF_REMAIN(writer->
binlog_buffer) < writer->cfg.max_record_size)
{
return sf_file_writer_flush(writer);
} else {
return 0;
}
}
int sf_file_writer_init(SFFileWriterInfo *writer, const char *data_path,
const char *subdir_name, const char *file_prefix,
const int max_record_size, const int buffer_size,
const int64_t file_rotate_size, const bool call_fsync)
{
int result;
int path_len;
bool create;
char filepath[PATH_MAX];
writer->total_count = 0;
writer->last_versions.pending = 0;
writer->last_versions.done = 0;
writer->flags = 0;
sf_file_writer_set_write_done_callback(writer, NULL, NULL);
if ((result=sf_binlog_buffer_init(&writer->
binlog_buffer, buffer_size)) != 0)
{
return result;
}
writer->cfg.max_record_size = max_record_size;
writer->cfg.call_fsync = call_fsync;
writer->cfg.file_rotate_size = file_rotate_size;
writer->cfg.data_path = data_path;
path_len = fc_combine_full_filepath(data_path, subdir_name, filepath);
if ((result=fc_check_mkdir_ex(filepath, 0775, &create)) != 0) {
return result;
}
if (create) {
SF_CHOWN_TO_RUNBY_RETURN_ON_ERROR(filepath);
}
writer->file.fd = -1;
fc_safe_strcpy(writer->cfg.subdir_name, subdir_name);
fc_safe_strcpy(writer->cfg.file_prefix, file_prefix);
writer->file.name.size = path_len + 32;
writer->file.name.str = (char *)fc_malloc(writer->file.name.size);
if (writer->file.name.str == NULL) {
return ENOMEM;
}
if ((result=get_binlog_index_from_file(writer)) != 0) {
return result;
}
if ((result=open_writable_binlog(writer)) != 0) {
return result;
}
return 0;
}
void sf_file_writer_destroy(SFFileWriterInfo *writer)
{
if (writer->file.fd >= 0) {
close(writer->file.fd);
writer->file.fd = -1;
}
if (writer->file.name.str != NULL) {
free(writer->file.name.str);
writer->file.name.str = NULL;
}
sf_binlog_buffer_destroy(&writer->binlog_buffer);
}
int sf_file_writer_set_indexes(SFFileWriterInfo *writer,
const int start_index, const int last_index)
{
int result;
if (writer->binlog.start_index != start_index ||
writer->binlog.last_index != last_index)
{
writer->binlog.start_index = start_index;
writer->binlog.last_index = last_index;
if ((result=write_to_binlog_index_file(writer)) != 0) {
return result;
}
}
return 0;
}
int sf_file_writer_set_binlog_start_index(SFFileWriterInfo *writer,
const int start_index)
{
int result;
if (writer->binlog.start_index != start_index) {
writer->binlog.start_index = start_index;
if ((result=write_to_binlog_index_file(writer)) != 0) {
return result;
}
}
return 0;
}
int sf_file_writer_set_binlog_write_index(SFFileWriterInfo *writer,
const int last_index)
{
int result;
if (writer->binlog.last_index != last_index) {
writer->binlog.last_index = last_index;
if ((result=write_to_binlog_index_file(writer)) != 0) {
return result;
}
}
return open_writable_binlog(writer);
}
int sf_file_writer_get_last_lines(const char *data_path,
const char *subdir_name, const int current_write_index,
char *buff, const int buff_size, int *count, int *length)
{
int result;
int target_count;
int count1;
char filename[PATH_MAX];
string_t lines;
target_count = *count;
sf_file_writer_get_filename(data_path, subdir_name,
current_write_index, filename, sizeof(filename));
if (access(filename, F_OK) == 0) {
if ((result=fc_get_last_lines(filename, buff, buff_size,
&lines, count)) != 0)
{
if (result != ENOENT) {
return result;
}
}
if (*count >= target_count || current_write_index == 0) {
memmove(buff, lines.str, lines.len);
*length = lines.len;
return 0;
}
} else {
result = errno != 0 ? errno : EPERM;
if (result == ENOENT) {
*count = 0;
*length = 0;
return 0;
} else {
logError("file: "__FILE__", line: %d, "
"stat file %s fail, errno: %d, error info: %s",
__LINE__, filename, result, STRERROR(result));
*count = 0;
*length = 0;
return result;
}
}
sf_file_writer_get_filename(data_path, subdir_name,
current_write_index - 1, filename, sizeof(filename));
if (access(filename, F_OK) != 0) {
result = errno != 0 ? errno : EPERM;
if (result == ENOENT) {
memmove(buff, lines.str, lines.len);
*length = lines.len;
return 0;
} else {
logError("file: "__FILE__", line: %d, "
"stat file %s fail, errno: %d, error info: %s",
__LINE__, filename, result, STRERROR(result));
*count = 0;
*length = 0;
return result;
}
}
count1 = target_count - *count;
if ((result=fc_get_last_lines(filename, buff,
buff_size, &lines, &count1)) != 0)
{
*count = 0;
*length = 0;
return result;
}
memmove(buff, lines.str, lines.len);
*length = lines.len;
if (*count == 0) {
*count = count1;
} else {
sf_file_writer_get_filename(data_path, subdir_name,
current_write_index, filename, sizeof(filename));
if ((result=fc_get_first_lines(filename, buff + (*length),
buff_size - (*length), &lines, count)) != 0)
{
*count = 0;
*length = 0;
return result;
}
*count += count1;
*length += lines.len;
}
return 0;
}
int sf_file_writer_get_last_line(const char *data_path,
const char *subdir_name, char *buff,
const int buff_size, int *length)
{
int result;
int last_index;
int count = 1;
if ((result=sf_file_writer_get_binlog_last_index(data_path,
subdir_name, &last_index)) != 0)
{
*length = 0;
return result;
}
return sf_file_writer_get_last_lines(data_path, subdir_name,
last_index, buff, buff_size, &count, length);
}
Reference in New Issue View Git Blame Copy Permalink