niklas/amduat
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases Wiki Activity

index-fs: append log records in place and add opt-in put precheck bypass

Browse source
This commit is contained in:
Carl Niklas Rydberg 2026-03-30 22:37:46 +02:00
parent 26fa4cce66
commit bc55ed994b
2 changed files with 329 additions and 203 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
include/amduat/asl/asl_store_index_fs.h
View file

@ -43,6 +43,10 @@ typedef struct {
uint64_t last_ingest_time_ns;
amduat_asl_snapshot_id_t next_snapshot_id;
bool snapshot_state_initialized;
bool log_tail_initialized;
uint64_t log_tail_logseq;
uint8_t log_tail_hash[32];
bool skip_exists_precheck;
void *open_segments;
pthread_mutex_t write_mutex;
uint32_t write_depth;

478
src/adapters/asl_store_index_fs/asl_store_index_fs.c
View file

@ -138,6 +138,19 @@ static amduat_asl_store_error_t amduat_asl_store_index_fs_put_indexed_impl(
amduat_reference_t *out_ref,
amduat_asl_index_state_t *out_state);
static bool amduat_asl_store_index_fs_env_truthy(const char *value) {
if (value == NULL || value[0] == '\0') {
return false;
}
if (strcmp(value, "1") == 0 || strcmp(value, "true") == 0 ||
strcmp(value, "TRUE") == 0 || strcmp(value, "yes") == 0 ||
strcmp(value, "YES") == 0 || strcmp(value, "on") == 0 ||
strcmp(value, "ON") == 0) {
return true;
}
return false;
}
static int amduat_asl_store_index_fs_lockfile_acquire(
amduat_asl_store_index_fs_t *fs) {
char lock_path[AMDUAT_ASL_STORE_INDEX_FS_ROOT_MAX + 32];
@ -194,6 +207,8 @@ static amduat_asl_store_error_t amduat_asl_store_index_fs_begin_write(
return AMDUAT_ASL_STORE_ERR_IO;
}
fs->write_lock_fd = lock_fd;
/* Another store instance may have appended while we were unlocked. */
fs->log_tail_initialized = false;
}
fs->write_depth += 1u;
return AMDUAT_ASL_STORE_OK;
@ -1193,28 +1208,6 @@ static amduat_asl_store_error_t amduat_asl_store_index_fs_load_log(
return AMDUAT_ASL_STORE_OK;
}
static amduat_asl_store_error_t amduat_asl_store_index_fs_write_log(
const char *log_path,
const char *log_dir,
amduat_asl_log_record_t *records,
size_t record_count) {
amduat_octets_t log_bytes;
amduat_asl_store_index_fs_write_status_t status;
if (!amduat_enc_asl_log_encode_v1(records, record_count, &log_bytes)) {
return AMDUAT_ASL_STORE_ERR_IO;
}
status = amduat_asl_store_index_fs_write_replace(log_dir, log_path,
log_bytes.data,
log_bytes.len);
amduat_octets_free(&log_bytes);
if (status == AMDUAT_ASL_STORE_INDEX_FS_WRITE_OK) {
return AMDUAT_ASL_STORE_OK;
}
return AMDUAT_ASL_STORE_ERR_IO;
}
static bool amduat_asl_store_index_fs_encode_artifact_ref(
amduat_reference_t ref,
uint8_t *out,
@ -1250,6 +1243,240 @@ static bool amduat_asl_store_index_fs_encode_artifact_ref(
return true;
}
static amduat_asl_store_error_t amduat_asl_store_index_fs_load_log_tail(
const char *log_path,
uint64_t *out_last_logseq,
uint8_t out_last_hash[AMDUAT_ASL_STORE_INDEX_FS_LOG_HASH_LEN]) {
int fd;
uint8_t *header_bytes = NULL;
size_t header_size = 0u;
uint64_t last_logseq = 0u;
uint8_t prev_hash[AMDUAT_ASL_STORE_INDEX_FS_LOG_HASH_LEN];
bool seen_record = false;
if (log_path == NULL || out_last_logseq == NULL || out_last_hash == NULL) {
return AMDUAT_ASL_STORE_ERR_IO;
}
memset(prev_hash, 0, sizeof(prev_hash));
fd = open(log_path, O_RDONLY);
if (fd < 0) {
if (errno == ENOENT || errno == ENOTDIR) {
*out_last_logseq = 0u;
memcpy(out_last_hash, prev_hash, sizeof(prev_hash));
return AMDUAT_ASL_STORE_OK;
}
return AMDUAT_ASL_STORE_ERR_IO;
}
if (!amduat_asl_store_index_fs_log_read_header_bytes(fd,
&header_bytes,
&header_size)) {
close(fd);
return AMDUAT_ASL_STORE_ERR_INTEGRITY;
}
free(header_bytes);
(void)header_size;
while (true) {
uint8_t record_header[16];
uint64_t logseq;
uint32_t record_type;
uint32_t payload_len;
uint8_t *payload = NULL;
uint8_t record_hash[AMDUAT_ASL_STORE_INDEX_FS_LOG_HASH_LEN];
uint8_t expected_hash[AMDUAT_ASL_STORE_INDEX_FS_LOG_HASH_LEN];
int rc;
rc = amduat_asl_store_index_fs_log_read_exact_eof(fd,
record_header,
sizeof(record_header));
if (rc == 0) {
break;
}
if (rc < 0) {
close(fd);
return AMDUAT_ASL_STORE_ERR_INTEGRITY;
}
logseq = amduat_asl_store_index_fs_load_u64_le(record_header);
if (seen_record && logseq <= last_logseq) {
close(fd);
return AMDUAT_ASL_STORE_ERR_INTEGRITY;
}
record_type =
amduat_asl_store_index_fs_log_load_u32_le(record_header + 8u);
payload_len =
amduat_asl_store_index_fs_log_load_u32_le(record_header + 12u);
if (payload_len != 0u) {
payload = (uint8_t *)malloc(payload_len);
if (payload == NULL) {
close(fd);
return AMDUAT_ASL_STORE_ERR_IO;
}
if (!amduat_asl_store_index_fs_log_read_exact(fd,
payload,
payload_len)) {
free(payload);
close(fd);
return AMDUAT_ASL_STORE_ERR_INTEGRITY;
}
}
if (!amduat_asl_store_index_fs_log_read_exact(fd,
record_hash,
sizeof(record_hash))) {
free(payload);
close(fd);
return AMDUAT_ASL_STORE_ERR_INTEGRITY;
}
if (!amduat_asl_store_index_fs_log_hash_record(prev_hash,
logseq,
record_type,
payload_len,
payload,
expected_hash) ||
memcmp(expected_hash, record_hash, sizeof(record_hash)) != 0) {
free(payload);
close(fd);
return AMDUAT_ASL_STORE_ERR_INTEGRITY;
}
memcpy(prev_hash, record_hash, sizeof(prev_hash));
last_logseq = logseq;
seen_record = true;
free(payload);
}
close(fd);
*out_last_logseq = last_logseq;
memcpy(out_last_hash, prev_hash, sizeof(prev_hash));
return AMDUAT_ASL_STORE_OK;
}
static amduat_asl_store_error_t amduat_asl_store_index_fs_append_log_record_file(
const char *index_path,
const char *log_path,
uint64_t prev_logseq,
const uint8_t prev_hash[AMDUAT_ASL_STORE_INDEX_FS_LOG_HASH_LEN],
uint32_t record_type,
const uint8_t *payload,
size_t payload_len,
uint64_t *out_new_logseq,
uint8_t out_new_hash[AMDUAT_ASL_STORE_INDEX_FS_LOG_HASH_LEN]) {
int fd = -1;
struct stat st;
bool needs_header = false;
uint8_t *header_bytes = NULL;
size_t header_size = 0u;
uint8_t record_header[16];
uint8_t record_hash[AMDUAT_ASL_STORE_INDEX_FS_LOG_HASH_LEN];
uint64_t new_logseq;
if (index_path == NULL || log_path == NULL || prev_hash == NULL ||
out_new_logseq == NULL || out_new_hash == NULL) {
return AMDUAT_ASL_STORE_ERR_IO;
}
if (payload_len != 0u && payload == NULL) {
return AMDUAT_ASL_STORE_ERR_IO;
}
if (payload_len > UINT32_MAX || prev_logseq == UINT64_MAX) {
return AMDUAT_ASL_STORE_ERR_UNSUPPORTED;
}
new_logseq = prev_logseq + 1u;
fd = open(log_path, O_RDWR | O_CREAT, 0644);
if (fd < 0) {
return AMDUAT_ASL_STORE_ERR_IO;
}
if (fstat(fd, &st) != 0) {
close(fd);
return AMDUAT_ASL_STORE_ERR_IO;
}
needs_header = st.st_size == 0;
if (!needs_header) {
if (lseek(fd, 0, SEEK_SET) < 0) {
close(fd);
return AMDUAT_ASL_STORE_ERR_IO;
}
if (!amduat_asl_store_index_fs_log_read_header_bytes(fd,
&header_bytes,
&header_size)) {
close(fd);
return AMDUAT_ASL_STORE_ERR_INTEGRITY;
}
free(header_bytes);
(void)header_size;
}
if (lseek(fd, 0, SEEK_END) < 0) {
close(fd);
return AMDUAT_ASL_STORE_ERR_IO;
}
if (needs_header) {
uint8_t header[AMDUAT_ASL_STORE_INDEX_FS_LOG_HEADER_LEN];
memcpy(header,
k_amduat_asl_store_index_fs_log_magic,
AMDUAT_ASL_STORE_INDEX_FS_LOG_MAGIC_LEN);
amduat_asl_store_index_fs_log_store_u32_le(
header + AMDUAT_ASL_STORE_INDEX_FS_LOG_MAGIC_LEN,
AMDUAT_ASL_STORE_INDEX_FS_LOG_VERSION);
amduat_asl_store_index_fs_log_store_u32_le(
header + AMDUAT_ASL_STORE_INDEX_FS_LOG_MAGIC_LEN + 4u,
AMDUAT_ASL_STORE_INDEX_FS_LOG_HEADER_LEN);
amduat_asl_store_index_fs_log_store_u64_le(
header + AMDUAT_ASL_STORE_INDEX_FS_LOG_MAGIC_LEN + 8u,
0u);
if (!amduat_asl_store_index_fs_log_write_exact(fd, header, sizeof(header))) {
close(fd);
return AMDUAT_ASL_STORE_ERR_IO;
}
}
if (!amduat_asl_store_index_fs_log_hash_record(prev_hash,
new_logseq,
record_type,
(uint32_t)payload_len,
payload,
record_hash)) {
close(fd);
return AMDUAT_ASL_STORE_ERR_IO;
}
amduat_asl_store_index_fs_log_store_u64_le(record_header, new_logseq);
amduat_asl_store_index_fs_log_store_u32_le(record_header + 8u, record_type);
amduat_asl_store_index_fs_log_store_u32_le(record_header + 12u,
(uint32_t)payload_len);
if (!amduat_asl_store_index_fs_log_write_exact(fd,
record_header,
sizeof(record_header)) ||
(payload_len != 0u &&
!amduat_asl_store_index_fs_log_write_exact(fd, payload, payload_len)) ||
!amduat_asl_store_index_fs_log_write_exact(fd,
record_hash,
sizeof(record_hash))) {
close(fd);
return AMDUAT_ASL_STORE_ERR_IO;
}
if (fsync(fd) != 0) {
close(fd);
return AMDUAT_ASL_STORE_ERR_IO;
}
if (close(fd) != 0) {
return AMDUAT_ASL_STORE_ERR_IO;
}
if (!amduat_asl_store_index_fs_fsync_directory(index_path)) {
return AMDUAT_ASL_STORE_ERR_IO;
}
*out_new_logseq = new_logseq;
memcpy(out_new_hash, record_hash, sizeof(record_hash));
return AMDUAT_ASL_STORE_OK;
}
static amduat_asl_store_error_t amduat_asl_store_index_fs_append_log_record_unlocked(
amduat_asl_store_index_fs_t *fs,
uint32_t record_type,
@ -1258,11 +1485,8 @@ static amduat_asl_store_error_t amduat_asl_store_index_fs_append_log_record_unlo
amduat_asl_index_state_t *out_state) {
char *index_path = NULL;
char *log_path = NULL;
amduat_asl_log_record_t *log_records = NULL;
size_t log_count = 0u;
amduat_asl_log_record_t *next = NULL;
uint64_t new_logseq;
uint8_t *payload_copy = NULL;
uint64_t new_logseq = 0u;
uint8_t new_hash[AMDUAT_ASL_STORE_INDEX_FS_LOG_HASH_LEN];
amduat_asl_store_error_t err;
if (fs == NULL) {
@ -1287,74 +1511,38 @@ static amduat_asl_store_error_t amduat_asl_store_index_fs_append_log_record_unlo
return AMDUAT_ASL_STORE_ERR_IO;
}
err = amduat_asl_store_index_fs_load_log(log_path, &log_records, &log_count);
if (!fs->log_tail_initialized) {
err = amduat_asl_store_index_fs_load_log_tail(log_path,
&fs->log_tail_logseq,
fs->log_tail_hash);
if (err != AMDUAT_ASL_STORE_OK) {
amduat_asl_store_index_fs_log_free(log_records, log_count);
free(log_path);
free(index_path);
return err;
}
if (log_count != 0u && log_records == NULL) {
free(log_path);
free(index_path);
return AMDUAT_ASL_STORE_ERR_IO;
fs->log_tail_initialized = true;
}
if (log_count == 0u) {
log_records = (amduat_asl_log_record_t *)calloc(1u,
sizeof(*log_records));
if (log_records == NULL) {
free(log_path);
free(index_path);
return AMDUAT_ASL_STORE_ERR_IO;
}
new_logseq = 1u;
} else {
if (log_records[log_count - 1u].logseq == UINT64_MAX) {
amduat_asl_store_index_fs_log_free(log_records, log_count);
free(log_path);
free(index_path);
return AMDUAT_ASL_STORE_ERR_IO;
}
new_logseq = log_records[log_count - 1u].logseq + 1u;
next = (amduat_asl_log_record_t *)realloc(
log_records, (log_count + 1u) * sizeof(*log_records));
if (next == NULL) {
amduat_asl_store_index_fs_log_free(log_records, log_count);
free(log_path);
free(index_path);
return AMDUAT_ASL_STORE_ERR_IO;
}
log_records = next;
}
if (payload_len != 0u) {
payload_copy = (uint8_t *)malloc(payload_len);
if (payload_copy == NULL) {
amduat_asl_store_index_fs_log_free(log_records, log_count);
free(log_path);
free(index_path);
return AMDUAT_ASL_STORE_ERR_IO;
}
memcpy(payload_copy, payload, payload_len);
}
log_records[log_count].logseq = new_logseq;
log_records[log_count].record_type = record_type;
log_records[log_count].payload = amduat_octets(payload_copy, payload_len);
memset(log_records[log_count].record_hash, 0,
sizeof(log_records[log_count].record_hash));
log_count += 1u;
err = amduat_asl_store_index_fs_write_log(log_path, index_path,
log_records, log_count);
amduat_asl_store_index_fs_log_free(log_records, log_count);
err = amduat_asl_store_index_fs_append_log_record_file(
index_path,
log_path,
fs->log_tail_logseq,
fs->log_tail_hash,
record_type,
payload,
payload_len,
&new_logseq,
new_hash);
free(log_path);
free(index_path);
if (err != AMDUAT_ASL_STORE_OK) {
return err;
}
fs->log_tail_logseq = new_logseq;
memcpy(fs->log_tail_hash, new_hash, sizeof(new_hash));
fs->log_tail_initialized = true;
if (out_state != NULL &&
!amduat_asl_store_index_fs_current_state_impl(fs, out_state)) {
return AMDUAT_ASL_STORE_ERR_IO;
@ -4114,6 +4302,7 @@ static amduat_asl_store_error_t amduat_asl_store_index_fs_put_indexed_unlocked_i
amduat_reference_t derived_ref;
amduat_octets_t artifact_bytes;
amduat_asl_index_state_t current_state;
bool has_current_state = false;
amduat_artifact_t existing_artifact;
amduat_asl_store_error_t err;
char *index_path;
@ -4121,7 +4310,6 @@ static amduat_asl_store_error_t amduat_asl_store_index_fs_put_indexed_unlocked_i
char *blocks_path;
char *segment_path;
char *block_path;
char *log_path;
uint16_t shard_id;
uint64_t segment_id;
uint64_t local_segment_id;
@ -4130,11 +4318,8 @@ static amduat_asl_store_error_t amduat_asl_store_index_fs_put_indexed_unlocked_i
amduat_asl_extent_record_t extent;
amduat_octets_t segment_bytes;
uint8_t segment_hash[AMDUAT_ASL_STORE_INDEX_FS_SEGMENT_HASH_LEN];
amduat_asl_log_record_t *log_records;
size_t log_count;
uint8_t *seal_payload;
size_t seal_len;
uint64_t new_logseq;
amduat_asl_store_index_fs_write_status_t write_status;
size_t artifact_len;
@ -4163,11 +4348,13 @@ static amduat_asl_store_error_t amduat_asl_store_index_fs_put_indexed_unlocked_i
fs->shard_count);
}
if (!fs->skip_exists_precheck) {
if (!amduat_asl_store_index_fs_current_state_impl(ctx, &current_state)) {
amduat_reference_free(&derived_ref);
amduat_octets_free(&artifact_bytes);
return AMDUAT_ASL_STORE_ERR_IO;
}
has_current_state = true;
existing_artifact = amduat_artifact(amduat_octets(NULL, 0u));
err = amduat_asl_store_index_fs_get_indexed_impl(ctx, derived_ref,
@ -4187,6 +4374,7 @@ static amduat_asl_store_error_t amduat_asl_store_index_fs_put_indexed_unlocked_i
amduat_octets_free(&artifact_bytes);
return err;
}
}
if (fs->segment_policy.small_artifact_threshold != 0u &&
artifact_bytes.len <= fs->segment_policy.small_artifact_threshold &&
@ -4205,6 +4393,12 @@ static amduat_asl_store_error_t amduat_asl_store_index_fs_put_indexed_unlocked_i
amduat_asl_store_index_fs_open_segment_contains(open_segment,
derived_ref)) {
*out_ref = derived_ref;
if (!has_current_state &&
!amduat_asl_store_index_fs_current_state_impl(ctx, &current_state)) {
amduat_octets_free(&artifact_bytes);
return AMDUAT_ASL_STORE_ERR_IO;
}
has_current_state = true;
amduat_asl_store_index_fs_fill_index_state(out_state,
current_state.snapshot_id,
current_state.log_position);
@ -4254,6 +4448,13 @@ static amduat_asl_store_error_t amduat_asl_store_index_fs_put_indexed_unlocked_i
return err;
}
} else {
if (!has_current_state &&
!amduat_asl_store_index_fs_current_state_impl(ctx, &current_state)) {
amduat_reference_free(&derived_ref);
amduat_octets_free(&artifact_bytes);
return AMDUAT_ASL_STORE_ERR_IO;
}
has_current_state = true;
amduat_asl_store_index_fs_fill_index_state(out_state,
current_state.snapshot_id,
current_state.log_position);
@ -4470,75 +4671,11 @@ static amduat_asl_store_error_t amduat_asl_store_index_fs_put_indexed_unlocked_i
return AMDUAT_ASL_STORE_ERR_IO;
}
if (!amduat_asl_store_index_fs_layout_build_log_path(fs->root_path,
&log_path)) {
amduat_reference_free(&derived_ref);
amduat_octets_free(&artifact_bytes);
free(index_path);
free(segments_path);
free(blocks_path);
return AMDUAT_ASL_STORE_ERR_IO;
}
log_records = NULL;
log_count = 0u;
err = amduat_asl_store_index_fs_load_log(log_path, &log_records, &log_count);
if (err != AMDUAT_ASL_STORE_OK) {
amduat_reference_free(&derived_ref);
amduat_octets_free(&artifact_bytes);
amduat_asl_store_index_fs_log_free(log_records, log_count);
free(log_path);
free(index_path);
free(segments_path);
free(blocks_path);
return err;
}
if (log_count != 0u && log_records == NULL) {
amduat_reference_free(&derived_ref);
amduat_octets_free(&artifact_bytes);
free(log_path);
free(index_path);
free(segments_path);
free(blocks_path);
return AMDUAT_ASL_STORE_ERR_IO;
}
if (log_count == 0u) {
log_records = (amduat_asl_log_record_t *)calloc(1u,
sizeof(*log_records));
if (log_records == NULL) {
amduat_reference_free(&derived_ref);
amduat_octets_free(&artifact_bytes);
free(log_path);
free(index_path);
free(segments_path);
free(blocks_path);
return AMDUAT_ASL_STORE_ERR_IO;
}
} else {
amduat_asl_log_record_t *next = (amduat_asl_log_record_t *)realloc(
log_records, (log_count + 1u) * sizeof(*log_records));
if (next == NULL) {
amduat_reference_free(&derived_ref);
amduat_octets_free(&artifact_bytes);
amduat_enc_asl_log_free(log_records, log_count);
free(log_path);
free(index_path);
free(segments_path);
free(blocks_path);
return AMDUAT_ASL_STORE_ERR_IO;
}
log_records = next;
}
seal_len = 8u + AMDUAT_ASL_STORE_INDEX_FS_SEGMENT_HASH_LEN;
seal_payload = (uint8_t *)malloc(seal_len);
if (seal_payload == NULL) {
amduat_reference_free(&derived_ref);
amduat_octets_free(&artifact_bytes);
amduat_asl_store_index_fs_log_free(log_records, log_count);
free(log_path);
free(index_path);
free(segments_path);
free(blocks_path);
@ -4554,34 +4691,13 @@ static amduat_asl_store_error_t amduat_asl_store_index_fs_put_indexed_unlocked_i
seal_payload[6] = (uint8_t)((segment_id >> 48) & 0xffu);
seal_payload[7] = (uint8_t)((segment_id >> 56) & 0xffu);
memcpy(seal_payload + 8, segment_hash, sizeof(segment_hash));
if (log_count == 0u) {
new_logseq = 1u;
} else {
if (log_records[log_count - 1u].logseq == UINT64_MAX) {
amduat_reference_free(&derived_ref);
amduat_octets_free(&artifact_bytes);
amduat_asl_store_index_fs_log_free(log_records, log_count);
free(log_path);
free(index_path);
free(segments_path);
free(blocks_path);
err = amduat_asl_store_index_fs_append_log_record_unlocked(
fs,
AMDUAT_ASL_LOG_RECORD_SEGMENT_SEAL,
seal_payload,
seal_len,
NULL);
free(seal_payload);
return AMDUAT_ASL_STORE_ERR_IO;
}
new_logseq = log_records[log_count - 1u].logseq + 1u;
}
log_records[log_count].logseq = new_logseq;
log_records[log_count].record_type = AMDUAT_ASL_LOG_RECORD_SEGMENT_SEAL;
log_records[log_count].payload = amduat_octets(seal_payload, seal_len);
memset(log_records[log_count].record_hash, 0,
sizeof(log_records[log_count].record_hash));
log_count += 1u;
err = amduat_asl_store_index_fs_write_log(log_path, index_path,
log_records, log_count);
amduat_asl_store_index_fs_log_free(log_records, log_count);
free(log_path);
free(index_path);
free(segments_path);
free(blocks_path);
@ -4596,7 +4712,7 @@ static amduat_asl_store_error_t amduat_asl_store_index_fs_put_indexed_unlocked_i
*out_ref = derived_ref;
amduat_asl_store_index_fs_fill_index_state(out_state,
current_state.snapshot_id,
new_logseq);
fs->log_tail_logseq);
amduat_asl_store_index_fs_update_ingest_state(fs, artifact_len);
amduat_asl_store_index_fs_maybe_snapshot_size(fs);
return AMDUAT_ASL_STORE_OK;
@ -4800,6 +4916,7 @@ amduat_asl_store_error_t amduat_asl_store_index_fs_snapshot_create(
free(manifest_path);
return err;
}
fs->log_tail_initialized = false;
{
amduat_asl_snapshot_manifest_t verify_manifest;
@ -5224,6 +5341,11 @@ bool amduat_asl_store_index_fs_init(amduat_asl_store_index_fs_t *fs,
fs->last_ingest_time_ns = 0u;
fs->next_snapshot_id = 0u;
fs->snapshot_state_initialized = false;
fs->log_tail_initialized = false;
fs->log_tail_logseq = 0u;
memset(fs->log_tail_hash, 0, sizeof(fs->log_tail_hash));
fs->skip_exists_precheck = amduat_asl_store_index_fs_env_truthy(
getenv("AMDUAT_ASL_INDEX_FS_SKIP_EXISTS_PRECHECK"));
fs->open_segments = NULL;
fs->write_depth = 0u;
fs->write_lock_fd = -1;