/*
* Authors:
* Dmitriy A. Gearasimov <gerasimov.dmitriy@demlabs.net>
* DeM Labs Inc. https://demlabs.net
* Kelvin Project https://github.com/kelvinblockchain
* Copyright (c) 2017-2019
* All rights reserved.
This file is part of DAP (Distributed Applications Platform) the open source project
DAP (Distributed Applications Platform) is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
DAP 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. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with any DAP based project. If not, see <http://www.gnu.org/licenses/>.
*/
#include "uthash.h"
#include <unistd.h>
#include "dap_chain.h"
#include "dap_chain_cell.h"
#include "dap_chain_cs.h"
#include "dap_common.h"
#include "dap_config.h"
#include "dap_strfuncs.h"
#include "dap_file_utils.h"
#ifdef DAP_OS_WINDOWS
#include <winternl.h>
#else
#include <sys/mman.h>
#endif
#define LOG_TAG "dap_chain_cell"
#define DAP_CHAIN_CELL_FILE_VERSION 1
#define DAP_CHAIN_CELL_FILE_SIGNATURE 0xfa340bef153eba48
#define DAP_CHAIN_CELL_FILE_TYPE_RAW 0
#define DAP_CHAIN_CELL_FILE_TYPE_COMPRESSED 1
#define DAP_MAPPED_VOLUME_LIMIT ( 1 << 28 ) // 256 MB for now, may be should be configurable?
/**
* @struct dap_chain_cell_file_header
*/
typedef struct dap_chain_cell_file_header
{
uint64_t signature;
uint32_t version;
uint8_t type;
dap_chain_id_t chain_id;
dap_chain_net_id_t chain_net_id;
dap_chain_cell_id_t cell_id;
} DAP_ALIGN_PACKED dap_chain_cell_file_header_t;
#ifdef DAP_OS_WINDOWS
typedef NTSTATUS (*pfn_NtCreateSection)(
OUT PHANDLE SectionHandle, IN ACCESS_MASK DesiredAccess,
IN OPTIONAL POBJECT_ATTRIBUTES ObjectAttributes,
IN OPTIONAL PLARGE_INTEGER MaximumSize, IN ULONG SectionPageProtection,
IN ULONG AllocationAttributes, IN OPTIONAL HANDLE FileHandle);
static pfn_NtCreateSection pfnNtCreateSection;
typedef enum _SECTION_INHERIT { ViewShare = 1, ViewUnmap = 2 } SECTION_INHERIT;
typedef NTSTATUS (*pfn_NtMapViewOfSection) (
IN HANDLE SectionHandle, IN HANDLE ProcessHandle, IN OUT PVOID *BaseAddress,
IN ULONG_PTR ZeroBits, IN SIZE_T CommitSize,
IN OUT OPTIONAL PLARGE_INTEGER SectionOffset, IN OUT PSIZE_T ViewSize,
IN SECTION_INHERIT InheritDisposition, IN ULONG AllocationType,
IN ULONG Win32Protect);
static pfn_NtMapViewOfSection pfnNtMapViewOfSection;
typedef NTSTATUS (*pfn_NtUnmapViewOfSection) (
IN HANDLE ProcessHandle, IN PVOID BaseAddress);
static pfn_NtUnmapViewOfSection pfnNtUnmapViewOfSection;
typedef NTSTATUS (*pfn_NtExtendSection) (
IN HANDLE SectionHandle, IN PLARGE_INTEGER NewSectionSize);
static pfn_NtExtendSection pfnNtExtendSection;
#endif
static bool s_debug_more = false;
/**
* @brief dap_chain_cell_init
* current version simply returns 0
* @return
*/
int dap_chain_cell_init(void)
{
s_debug_more = dap_config_get_item_bool_default(g_config, "chain", "debug_more", false);
#ifdef DAP_OS_WINDOWS
if ( dap_config_get_item_bool_default(g_config, "ledger", "mapped", true) ) {
HMODULE ntdll = GetModuleHandle("ntdll.dll");
if ( !ntdll )
return log_it(L_CRITICAL, "Ntdll error"), -1;
pfnNtCreateSection = (pfn_NtCreateSection) GetProcAddress(ntdll, "NtCreateSection");
pfnNtMapViewOfSection = (pfn_NtMapViewOfSection) GetProcAddress(ntdll, "NtMapViewOfSection");
pfnNtExtendSection = (pfn_NtExtendSection) GetProcAddress(ntdll, "NtExtendSection");
pfnNtUnmapViewOfSection = (pfn_NtUnmapViewOfSection)GetProcAddress(ntdll, "NtUnmapViewOfSection");
}
#endif
//s_cells_path = dap_config_get_item_str(g_config,"resources","cells_storage");
return 0;
}
DAP_STATIC_INLINE int s_cell_file_write_header(dap_chain_cell_t *a_cell)
{
dap_chain_cell_file_header_t l_hdr = {
.signature = DAP_CHAIN_CELL_FILE_SIGNATURE,
.version = DAP_CHAIN_CELL_FILE_VERSION,
.type = DAP_CHAIN_CELL_FILE_TYPE_RAW,
.chain_id = a_cell->chain->id,
.chain_net_id = a_cell->chain->net_id,
.cell_id = a_cell->id
};
return (int)fwrite(&l_hdr, sizeof(l_hdr), 1, a_cell->file_storage);
}
DAP_STATIC_INLINE int s_cell_map_new_volume(dap_chain_cell_t *a_cell, size_t a_fpos) {
int l_fildes = fileno(a_cell->file_storage);
#ifdef DAP_OS_WINDOWS
HANDLE hSection = NULL;
if ( !a_fpos ) {
//if (a_cell->map_range_bounds)
// NtClose( (HANDLE)a_cell->map_range_bounds->data );
LARGE_INTEGER SectionSize = {
.QuadPart = ( fseek(a_cell->file_storage, 0, SEEK_END), ftell(a_cell->file_storage) )
};
NTSTATUS err = pfnNtCreateSection(&hSection, SECTION_MAP_READ|SECTION_EXTEND_SIZE|SECTION_MAP_WRITE,
NULL, &SectionSize, PAGE_READWRITE, SEC_RESERVE,
(HANDLE)_get_osfhandle(l_fildes));
if ( !NT_SUCCESS(err) ) {
log_it(L_ERROR, "NtCreateSection() failed, status %lx", err);
return -1;
}
a_cell->map_range_bounds = dap_list_append(a_cell->map_range_bounds, hSection);
}
#endif
size_t l_map_size = dap_page_roundup(DAP_MAPPED_VOLUME_LIMIT),
#ifdef DAP_OS_WINDOWS
l_volume_start = a_fpos ? dap_64k_rounddown(a_fpos) : 0,
#else
l_volume_start = a_fpos ? dap_page_rounddown(a_fpos) : 0,
#endif
l_offset = a_fpos - l_volume_start;
#ifdef DAP_OS_WINDOWS
hSection = (HANDLE)a_cell->map_range_bounds->data;
a_cell->map = NULL;
int err = 0;
LARGE_INTEGER Offset = {
.QuadPart = l_volume_start
};
if ( !NT_SUCCESS ( err = pfnNtMapViewOfSection(hSection, GetCurrentProcess(),
(HANDLE)&a_cell->map, 0, 0,
&Offset, &l_map_size,
ViewUnmap, MEM_RESERVE,
PAGE_WRITECOPY) ) )
{
log_it(L_ERROR, "NtMapViewOfSection() failed, status %lx", err);
NtClose(hSection);
return -1;
}
#else
if ( MAP_FAILED == (a_cell->map = mmap(NULL, l_map_size, PROT_READ|PROT_WRITE,
MAP_PRIVATE, fileno(a_cell->file_storage), l_volume_start)) ) {
log_it(L_ERROR, "Chain cell \"%s\" 0x%016"DAP_UINT64_FORMAT_X" cannot be mapped, errno %d",
a_cell->file_storage_path, a_cell->id.uint64, errno);
return -1;
}
#endif
a_cell->map_pos = a_cell->map + l_offset;
a_cell->map_range_bounds = dap_list_append(a_cell->map_range_bounds, a_cell->map);
a_cell->map_range_bounds = dap_list_append(a_cell->map_range_bounds, a_cell->map_end = a_cell->map + l_map_size);
return 0;
}
/**
* @brief dap_chain_cell_find_by_id
* get dap_chain_cell_t object by cell (shard) id
* @param a_chain dap_chain_t object
* @param a_cell_id dap_chain_cell_id_t cell (shard) id
* @return dap_chain_cell_t*
*/
dap_chain_cell_t * dap_chain_cell_find_by_id(dap_chain_t * a_chain, dap_chain_cell_id_t a_cell_id)
{
if (!a_chain->cells)
return NULL;
dap_chain_cell_t *l_cell = NULL;
pthread_rwlock_rdlock(&a_chain->cell_rwlock);
HASH_FIND(hh, a_chain->cells, &a_cell_id, sizeof(dap_chain_cell_id_t), l_cell);
pthread_rwlock_unlock(&a_chain->cell_rwlock);
return l_cell;
}
/**
* @brief
* a_cell_id if < 0 then not used
* @param a_chain dap_chain_t object
* @param a_cell_id dap_chain_cell_id_t cell (shard) id
* @return dap_chain_cell_t*
*/
dap_chain_cell_t * dap_chain_cell_create_fill(dap_chain_t * a_chain, dap_chain_cell_id_t a_cell_id)
{
dap_chain_cell_t * l_cell = NULL;
pthread_rwlock_wrlock(&a_chain->cell_rwlock);
HASH_FIND(hh, a_chain->cells, &a_cell_id, sizeof(dap_chain_cell_id_t), l_cell);
if (l_cell) {
pthread_rwlock_unlock(&a_chain->cell_rwlock);
return l_cell;
}
#define CLEANUP_AND_RET return ({ fclose(l_file); DAP_DELETE(l_cell); pthread_rwlock_unlock(&a_chain->cell_rwlock); NULL; })
char file_storage_path[MAX_PATH];
snprintf(file_storage_path, MAX_PATH, "%s/%0"DAP_UINT64_FORMAT_x".dchaincell",
DAP_CHAIN_PVT(a_chain)->file_storage_dir, a_cell_id.uint64);
FILE *l_file = fopen(file_storage_path, "a+b");
if ( !l_file ) {
log_it(L_ERROR, "Chain cell \"%s\" 0x%016"DAP_UINT64_FORMAT_X" cannot be opened, error %d",
file_storage_path, a_cell_id.uint64, errno);
CLEANUP_AND_RET;
}
if ( !(l_cell = DAP_NEW_Z(dap_chain_cell_t)) )
CLEANUP_AND_RET;
*l_cell = (dap_chain_cell_t) {
.id = a_cell_id,
.chain = a_chain,
.file_storage = l_file,
.storage_rwlock = PTHREAD_RWLOCK_INITIALIZER
};
size_t l_size = ( fseek(l_file, 0, SEEK_END), ftell(l_file) );
if ( l_size < sizeof(dap_chain_cell_file_header_t) ) {
if (l_size) {
log_it(L_INFO, "Possibly corrupt cell storage 0x%016"DAP_UINT64_FORMAT_X" \"%s\", rewriting it",
a_cell_id.uint64, file_storage_path);
l_file = freopen(file_storage_path, "w+b", l_file);
}
if ( !s_cell_file_write_header(l_cell) ) {
log_it(L_ERROR, "Can't init file storage for cell 0x%016"DAP_UINT64_FORMAT_X" \"%s\", errno %d",
a_cell_id.uint64, file_storage_path, errno);
CLEANUP_AND_RET;
}
log_it(L_NOTICE, "Initialized file storage for cell 0x%016"DAP_UINT64_FORMAT_X" \"%s\"",
a_cell_id.uint64, file_storage_path);
fflush(l_file);
}
if ( a_chain->is_mapped && s_cell_map_new_volume(l_cell, 0) ) {
CLEANUP_AND_RET;
}
#undef CLEANUP_AND_RET
memcpy(l_cell->file_storage_path, file_storage_path, sizeof(file_storage_path));
debug_if (s_debug_more && a_chain->is_mapped, L_DEBUG, "Mapped volume size is %lu", (size_t)(l_cell->map_end - l_cell->map));
HASH_ADD(hh, a_chain->cells, id, sizeof(dap_chain_cell_id_t), l_cell);
pthread_rwlock_unlock(&a_chain->cell_rwlock);
return l_cell;
}
/**
* @brief
* close a_cell->file_storage file object
* @param a_cell dap_chain_cell_t object
*/
void dap_chain_cell_close(dap_chain_cell_t *a_cell)
{
if(!a_cell)
return;
if(a_cell->file_storage) {
fclose(a_cell->file_storage);
a_cell->file_storage = NULL;
}
if (a_cell->chain->is_mapped) {
dap_list_t *l_iter = a_cell->map_range_bounds;
#ifdef DAP_OS_WINDOWS
l_iter = l_iter->next;
#endif
for (; l_iter; l_iter = l_iter->next) {
if (l_iter->next) {
debug_if(s_debug_more, L_DEBUG, "Unmap volume %p (%lu bytes)", l_iter->data, (size_t)(l_iter->next->data - l_iter->data));
#ifdef DAP_OS_WINDOWS
pfnNtUnmapViewOfSection(GetCurrentProcess(), l_iter->data);
#else
munmap(l_iter->data, (size_t)(l_iter->next->data - l_iter->data));
#endif
l_iter = l_iter->next;
}
}
#ifdef DAP_OS_WINDOWS
NtClose(a_cell->map_range_bounds->data);
#endif
dap_list_free(a_cell->map_range_bounds);
}
#ifdef DAP_OS_WINDOWS
char *l_new = strstr(a_cell->file_storage_path, ".new");
if (l_new) {
char *l_orig = dap_strdup(a_cell->file_storage_path);
*l_new = '\0';
remove(a_cell->file_storage_path);
rename(l_orig, a_cell->file_storage_path);
DAP_DELETE(l_orig);
}
#endif
}
/**
* @brief
* free chain cell objects
* @param a_cell dap_chain_cell_t object
*/
void dap_chain_cell_delete(dap_chain_cell_t *a_cell)
{
if(!a_cell)
return;
dap_chain_cell_close(a_cell);
if (a_cell->chain->cells) {
dap_chain_cell_t *l_cell = NULL;
dap_chain_cell_id_t l_cell_id = {
.uint64 = a_cell->id.uint64
};
pthread_rwlock_wrlock(&a_cell->chain->cell_rwlock);
HASH_FIND(hh, a_cell->chain->cells, &l_cell_id, sizeof(dap_chain_cell_id_t), l_cell);
if (l_cell)
HASH_DEL(a_cell->chain->cells, l_cell);
pthread_rwlock_unlock(&a_cell->chain->cell_rwlock);
}
a_cell->chain = NULL;
a_cell->file_storage_path[0] = '\0';
pthread_rwlock_destroy(&a_cell->storage_rwlock);
DAP_DELETE(a_cell);
}
void dap_chain_cell_delete_all_and_free_file(dap_chain_t *a_chain) {
if (!a_chain)
return;
pthread_rwlock_wrlock(&a_chain->cell_rwlock);
dap_chain_cell_t *l_cell, *l_tmp;
HASH_ITER(hh, a_chain->cells, l_cell, l_tmp) {
char *l_fsp = dap_strdup(l_cell->file_storage_path);
dap_chain_cell_id_t l_cell_id = l_cell->id;
dap_chain_cell_close(l_cell);
dap_chain_cell_t * l_cell_nh = DAP_NEW_Z(dap_chain_cell_t);
FILE *l_file = fopen(l_fsp, "w+b");
if ( !l_file ) {
log_it(L_ERROR, "Chain cell \"%s\" 0x%016"DAP_UINT64_FORMAT_X" cannot be opened, error %d",
l_fsp, l_cell_id.uint64, errno);
}
*l_cell_nh = (dap_chain_cell_t) {
.id = l_cell_id,
.chain = a_chain,
.file_storage = l_file
};
if ( !s_cell_file_write_header(l_cell_nh)) {
log_it(L_ERROR, "Can't init file storage for cell 0x%016"DAP_UINT64_FORMAT_X" \"%s\", errno %d",
l_cell_id.uint64, l_fsp, errno);
} else {
log_it(L_NOTICE, "Reinitialized file storage for cell 0x%016"DAP_UINT64_FORMAT_X" \"%s\"",
l_cell_id.uint64, l_fsp);
}
dap_chain_cell_close(l_cell_nh);
DAP_DELETE(l_fsp);
HASH_DEL(a_chain->cells, l_cell);
pthread_rwlock_destroy(&l_cell->storage_rwlock);
DAP_DELETE(l_cell);
}
pthread_rwlock_unlock(&a_chain->cell_rwlock);
}
void dap_chain_cell_delete_all(dap_chain_t *a_chain) {
if (!a_chain)
return;
pthread_rwlock_wrlock(&a_chain->cell_rwlock);
dap_chain_cell_t *l_cell, *l_tmp;
HASH_ITER(hh, a_chain->cells, l_cell, l_tmp) {
dap_chain_cell_close(l_cell);
HASH_DEL(a_chain->cells, l_cell);
pthread_rwlock_destroy(&l_cell->storage_rwlock);
DAP_DELETE(l_cell);
}
pthread_rwlock_unlock(&a_chain->cell_rwlock);
}
/**
* @brief dap_chain_cell_load
* load cell file, which is pointed in a_cell_file_path variable, for example "0.dchaincell"
* @param a_chain dap_chain_t object
* @param a_cell_file_path contains name of chain, for example "0.dchaincell"
* @return
*/
int dap_chain_cell_load(dap_chain_t *a_chain, dap_chain_cell_t *a_cell)
{
if (!a_cell)
return -1;
size_t l_size = ( fseek(a_cell->file_storage, 0, SEEK_END), ftell(a_cell->file_storage) ), l_pos = 0;
if ( l_size < sizeof(dap_chain_cell_file_header_t) ) {
log_it(L_ERROR, "Chain cell \"%s\" is corrupt, create new file", a_cell->file_storage_path);
return -1;
}
int l_ret = 0;
dap_chain_cell_file_header_t *l_hdr = NULL;
if (a_chain->is_mapped) {
l_hdr = (dap_chain_cell_file_header_t*)a_cell->map;
} else {
fseek(a_cell->file_storage, 0, SEEK_SET);
l_hdr = DAP_NEW(dap_chain_cell_file_header_t);
if ( fread(l_hdr, 1, sizeof(*l_hdr), a_cell->file_storage) != sizeof(*l_hdr) ) {
log_it(L_ERROR,"Can't read chain header \"%s\"", a_cell->file_storage_path);
fclose(a_cell->file_storage);
DAP_DELETE(l_hdr);
return -2;
}
}
if (l_hdr->signature != DAP_CHAIN_CELL_FILE_SIGNATURE) {
log_it(L_ERROR, "Wrong signature in chain \"%s\", possible file corrupt", a_cell->file_storage_path);
fclose(a_cell->file_storage);
if (!a_chain->is_mapped) DAP_DELETE(l_hdr);
return -3;
}
if (l_hdr->version < DAP_CHAIN_CELL_FILE_VERSION ){
log_it(L_ERROR, "Too low chain version, backup files");
fclose(a_cell->file_storage);
if (!a_chain->is_mapped) DAP_DELETE(l_hdr);
return -4;
}
l_pos += sizeof(dap_chain_cell_file_header_t);
if (a_chain->is_mapped)
a_cell->map_pos += l_pos;
if (l_size == l_pos) {
fseek(a_cell->file_storage, l_pos, SEEK_SET);
return 0;
}
uint64_t q = 0;
if (a_chain->is_mapped) {
a_cell->map_pos = a_cell->map + sizeof(dap_chain_cell_file_header_t);
for ( uint64_t l_el_size = 0; l_pos < l_size; ++q, l_pos += l_el_size + sizeof(uint64_t), a_chain->load_progress = (int)((double)l_pos/l_size * 100 + 0.5)) {
size_t space_left = (size_t)( a_cell->map_end - a_cell->map_pos );
if ( space_left < sizeof(uint64_t) || (space_left - sizeof(uint64_t)) < *(uint64_t*)a_cell->map_pos )
if ( s_cell_map_new_volume(a_cell, l_pos) )
break;
l_el_size = *(uint64_t*)a_cell->map_pos;
dap_hash_fast_t l_atom_hash;
dap_chain_atom_ptr_t l_atom = (dap_chain_atom_ptr_t)(a_cell->map_pos += sizeof(uint64_t));
dap_hash_fast(l_atom, l_el_size, &l_atom_hash);
a_chain->callback_atom_add(a_chain, l_atom, l_el_size, &l_atom_hash);
a_cell->map_pos += l_el_size;
}
} else {
DAP_DELETE(l_hdr);
size_t l_read = 0;
uint64_t l_el_size = 0;
while ((l_read = fread(&l_el_size, 1, sizeof(l_el_size), a_cell->file_storage)) && !feof(a_cell->file_storage)) {
if (l_read != sizeof(l_el_size) || l_el_size == 0) {
log_it(L_ERROR, "Corrupted element size %zu, chain %s is damaged", l_el_size, a_cell->file_storage_path);
l_ret = -4;
break;
}
dap_chain_atom_ptr_t l_element = DAP_NEW_SIZE(dap_chain_atom_ptr_t, l_el_size);
if (!l_element) {
log_it(L_CRITICAL, "Memory allocation error");
l_ret = -5;
break;
}
l_pos += sizeof(uint64_t) + ( l_read = fread((void*)l_element, 1, l_el_size, a_cell->file_storage) );
a_chain->load_progress = (int)((double)l_pos/l_size * 100 + 0.5);
if (l_read != l_el_size) {
log_it(L_ERROR, "Read only %lu of %zu bytes, stop cell loading", l_read, l_el_size);
DAP_DELETE(l_element);
l_ret = -6;
break;
}
dap_hash_fast_t l_atom_hash = {};
dap_hash_fast(l_element, l_el_size, &l_atom_hash);
dap_chain_atom_verify_res_t l_res = a_chain->callback_atom_add(a_chain, l_element, l_el_size, &l_atom_hash);
if (l_res != ATOM_ACCEPT && l_res != ATOM_FORK) {
DAP_DELETE(l_element);
}
++q;
}
}
fseek(a_cell->file_storage, l_pos, SEEK_SET);
log_it(L_INFO, "Loaded %lu atoms in cell %s", q, a_cell->file_storage_path);
return l_ret;
}
static int s_cell_file_atom_add(dap_chain_cell_t *a_cell, dap_chain_atom_ptr_t a_atom, uint64_t a_atom_size)
{
if (!a_atom || !a_atom_size) {
log_it(L_CRITICAL, "Invalid arguments");
return -1;
}
if (a_cell->chain->is_mapped) {
size_t l_pos = ( fseek(a_cell->file_storage, 0, SEEK_END), ftell(a_cell->file_storage) );
debug_if (s_debug_more, L_DEBUG, "Before filling volume for atom size %lu, stream pos of %s is %lu, map pos is %lu, space left in map %lu",
a_atom_size, a_cell->file_storage_path, l_pos, (size_t)(a_cell->map_pos - a_cell->map), (size_t)(a_cell->map_end - a_cell->map_pos));
if ( a_atom_size + sizeof(uint64_t) > (size_t)(a_cell->map_end - a_cell->map_pos) )
if ( s_cell_map_new_volume(a_cell, l_pos) )
return -2;
}
debug_if (s_debug_more && a_cell->chain->is_mapped, L_DEBUG, "Before writing an atom of size %lu, stream pos of %s is %lu and pos is %lu, space left in map %lu",
a_atom_size, a_cell->file_storage_path, ftell(a_cell->file_storage),
(size_t)(a_cell->map_pos - a_cell->map), (size_t)(a_cell->map_end - a_cell->map_pos));
if (fwrite(&a_atom_size, sizeof(a_atom_size), 1, a_cell->file_storage) != 1) {
log_it (L_ERROR, "Can't write atom data size from cell 0x%016"DAP_UINT64_FORMAT_X" in \"%s\"",
a_cell->id.uint64,
a_cell->file_storage_path);
return -2;
}
if (fwrite(a_atom, a_atom_size, 1, a_cell->file_storage) != 1) {
log_it (L_ERROR, "Can't write data from cell 0x%016"DAP_UINT64_FORMAT_X" to the file \"%s\"",
a_cell->id.uint64,
a_cell->file_storage_path);
return -3;
}
debug_if (s_debug_more && a_cell->chain->is_mapped, L_DEBUG, "After writing an atom of size %lu, stream pos of %s is %lu and map shift is %lu",
a_atom_size, a_cell->file_storage_path, ftell(a_cell->file_storage),
(size_t)(a_cell->map_pos - a_cell->map));
return 0;
}
/**
* @brief s_cell_file_append
* add atoms to selected chain
* @param a_cell - cell object. Contains file path to cell storage data, for example - "0.dchaincell"
* a_cell->chain contains
* name - "zerochain"
* net_name - "kelvin-testnet"
* filepath - "C:\\Users\\Public\\Documents\\cellframe-node\\var\\lib\\network\\kelvin-testnet\\zerochain\\/0.dchaincell"
* @param a_atom
* @param a_atom_size
* @return
*/
ssize_t dap_chain_cell_file_append(dap_chain_cell_t *a_cell, const void *a_atom, size_t a_atom_size)
{
if(!a_cell)
return -1;
if (!a_atom && !a_cell->chain) {
log_it(L_WARNING,"Chain not found for cell 0x%016"DAP_UINT64_FORMAT_X" ( %s )",
a_cell->id.uint64, a_cell->file_storage_path);
return -1;
}
size_t l_total_res = 0, l_count = 0;
bool l_err = false;
pthread_rwlock_wrlock(&a_cell->storage_rwlock);
if (!a_atom || !a_atom_size) {
#ifdef DAP_OS_WINDOWS
strcat(a_cell->file_storage_path, ".new");
#endif
a_cell->file_storage = freopen(a_cell->file_storage_path, "w+b", a_cell->file_storage);
debug_if (s_debug_more,L_DEBUG, "Rewinding file %s", a_cell->file_storage_path);
bool was_mapped = a_cell->chain->is_mapped;
a_cell->chain->is_mapped = false;
if ( !s_cell_file_write_header(a_cell) ) {
log_it(L_ERROR, "Chain cell \"%s\" 0x%016"DAP_UINT64_FORMAT_X": can't fill header",
a_cell->file_storage_path, a_cell->id.uint64);
a_cell->chain->is_mapped = was_mapped;
pthread_rwlock_unlock(&a_cell->storage_rwlock);
return -2;
}
l_total_res += sizeof(dap_chain_cell_file_header_t);
dap_chain_atom_iter_t *l_atom_iter = a_cell->chain->callback_atom_iter_create(a_cell->chain, a_cell->id, NULL);
dap_chain_atom_ptr_t l_atom;
uint64_t l_atom_size = 0;
for (l_atom = a_cell->chain->callback_atom_iter_get(l_atom_iter, DAP_CHAIN_ITER_OP_FIRST, &l_atom_size);
l_atom && l_atom_size;
l_atom = a_cell->chain->callback_atom_iter_get(l_atom_iter, DAP_CHAIN_ITER_OP_NEXT, &l_atom_size))
{
if ( s_cell_file_atom_add(a_cell, l_atom, l_atom_size) ) {
l_err = true;
break;
} else {
l_total_res += sizeof(uint64_t) + l_atom_size;
++l_count;
}
}
a_cell->chain->is_mapped = was_mapped;
a_cell->chain->callback_atom_iter_delete(l_atom_iter);
debug_if (s_debug_more && a_cell->chain->is_mapped,L_DEBUG, "After rewriting file %s, stream pos is %lu and map pos is %lu",
a_cell->file_storage_path, ftell(a_cell->file_storage),
(size_t)(a_cell->map_pos - a_cell->map));
} else {
debug_if (s_debug_more && a_cell->chain->is_mapped,L_DEBUG, "Before appending an atom of size %lu, stream pos of %s is %lu, map pos is %lu",
a_atom_size, a_cell->file_storage_path, ftell(a_cell->file_storage),
(size_t)(a_cell->map_pos - a_cell->map));
if ( s_cell_file_atom_add(a_cell, a_atom, a_atom_size) ) {
log_it(L_ERROR, "Chain cell \"%s\" 0x%016"DAP_UINT64_FORMAT_X": can't save atom!",
a_cell->file_storage_path, a_cell->id.uint64);
pthread_rwlock_unlock(&a_cell->storage_rwlock);
return -4;
}
debug_if (s_debug_more && a_cell->chain->is_mapped, L_DEBUG,"After appending an atom of size %lu, stream pos of %s is %lu, map pos is %lu",
a_atom_size, a_cell->file_storage_path, ftell(a_cell->file_storage),
(size_t)(a_cell->map_end - a_cell->map_pos));
++l_count;
l_total_res = a_atom_size + sizeof(uint64_t);
}
if (l_total_res) {
fflush(a_cell->file_storage);
#ifdef DAP_OS_WINDOWS
if (a_cell->chain->is_mapped) {
LARGE_INTEGER SectionSize = (LARGE_INTEGER) { .QuadPart = ftell(a_cell->file_storage) };
HANDLE hSection = (HANDLE)a_cell->map_range_bounds->data;
NTSTATUS err = pfnNtExtendSection(hSection, &SectionSize);
if ( !NT_SUCCESS(err) ) {
log_it(L_ERROR, "NtExtendSection() failed, status %lx", err);
return -4;
}
}
#endif
log_it(L_DEBUG, "Chain cell \"%s\" 0x%016"DAP_UINT64_FORMAT_X": saved %zu atoms (%zu bytes)",
a_cell->file_storage_path, a_cell->id.uint64, l_count, l_total_res);
if (l_err) {
log_it(L_WARNING, "Not all data was saved due to writing error!");
}
} else {
log_it(L_ERROR, "Chain cell \"%s\" 0x%016"DAP_UINT64_FORMAT_X": nothing saved!",
a_cell->file_storage_path, a_cell->id.uint64);
}
pthread_rwlock_unlock(&a_cell->storage_rwlock);
return l_total_res;
}