cozis/ToastyFS
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add crash logger

Browse Source
This commit is contained in:
cozis
2025-12-04 12:25:19 +01:00
parent 5c192a1e52
commit e73bb3a7d8
7 changed files with 612 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+1
View File
@@ -3,6 +3,7 @@
*.o
*.a
chunk_server_data_*
crash_*.txt
# Coverage reports
coverage_report/
misc/PROTOCOL.txt
+1 -1
View File
@@ -208,7 +208,7 @@ The following is the list of messages the Metadata Server may send to a Client:
If the truncated field is non-zero, the actual item count for this directory is greater than the one sent in the message.
[ READ_ERROR | MS -> C ]
See CREATE_ERROR
TODO
[ READ_SUCCESS | MS -> C ]
src/client.c
+54 -13
View File
@@ -33,6 +33,8 @@ typedef pthread_mutex_t Mutex;
#define PARALLEL_LIMIT 5
#define CLIENT_TRACE(fmt, ...) fprintf(stderr, "CLIENT: " fmt "\n", ##__VA_ARGS__);
typedef struct {
SHA256 hash;
char* dst;
@@ -1465,30 +1467,49 @@ static void process_event_for_write(ToastyFS *toasty,
return;
}
if (type != MESSAGE_TYPE_READ_SUCCESS) {
toasty->operations[opidx].result = (ToastyResult) { .type=TOASTY_RESULT_WRITE_ERROR, .user=toasty->operations[opidx].user };
return;
}
if (!binary_read(&reader, NULL, sizeof(uint32_t))) {
toasty->operations[opidx].result = (ToastyResult) { .type=TOASTY_RESULT_WRITE_ERROR, .user=toasty->operations[opidx].user };
return;
}
// Read generation counter
uint64_t gen;
uint32_t chunk_size;
uint32_t num_hashes;
if (type != MESSAGE_TYPE_READ_SUCCESS) {
if ((toasty->operations[opidx].flags & TOASTY_WRITE_CREATE_IF_MISSING) == 0) {
toasty->operations[opidx].result = (ToastyResult) { .type=TOASTY_RESULT_WRITE_ERROR, .user=toasty->operations[opidx].user };
return;
}
uint16_t message_len;
if (!binary_read(&reader, &message_len, sizeof(message_len))) {
toasty->operations[opidx].result = (ToastyResult) { .type=TOASTY_RESULT_WRITE_ERROR, .user=toasty->operations[opidx].user };
return;
}
if (!binary_read(&reader, NULL, message_len)) {
toasty->operations[opidx].result = (ToastyResult) { .type=TOASTY_RESULT_WRITE_ERROR, .user=toasty->operations[opidx].user };
return;
}
gen = 0; // TODO: is setting the generation to 0 right?
chunk_size = 4096; // The creation flag defaults to a chunk size of 4K
num_hashes = 0;
} else {
// Read generation counter
if (!binary_read(&reader, &gen, sizeof(gen))) {
toasty->operations[opidx].result = (ToastyResult) { .type=TOASTY_RESULT_WRITE_ERROR, .user=toasty->operations[opidx].user };
return;
}
toasty->operations[opidx].expect_gen = gen;
uint32_t chunk_size;
if (!binary_read(&reader, &chunk_size, sizeof(chunk_size))) {
toasty->operations[opidx].result = (ToastyResult) { .type=TOASTY_RESULT_WRITE_ERROR, .user=toasty->operations[opidx].user };
return;
}
toasty->operations[opidx].chunk_size = chunk_size;
// Skip file_length field that comes after chunk_size
if (!binary_read(&reader, NULL, sizeof(uint32_t))) {
@@ -1496,11 +1517,14 @@ static void process_event_for_write(ToastyFS *toasty,
return;
}
uint32_t num_hashes;
if (!binary_read(&reader, &num_hashes, sizeof(num_hashes))) {
toasty->operations[opidx].result = (ToastyResult) { .type=TOASTY_RESULT_WRITE_ERROR, .user=toasty->operations[opidx].user };
return;
}
}
toasty->operations[opidx].expect_gen = gen;
toasty->operations[opidx].chunk_size = chunk_size;
uint32_t num_all_hasehs = (toasty->operations[opidx].len + chunk_size - 1) / chunk_size;
uint32_t num_new_hashes = num_all_hasehs - num_hashes;
@@ -2168,6 +2192,7 @@ int toasty_process_events(ToastyFS *toasty, void **contexts, struct pollfd *poll
switch (events[i].type) {
case EVENT_WAKEUP:
CLIENT_TRACE("TCP EVENT: wakeup");
// Do nothing
break;
@@ -2176,6 +2201,8 @@ int toasty_process_events(ToastyFS *toasty, void **contexts, struct pollfd *poll
int tag = tcp_get_tag(&toasty->tcp, conn_idx);
if (tag != TAG_METADATA_SERVER)
toasty->chunk_servers[tag].connected = true;
CLIENT_TRACE("TCP EVENT: CONNECT (tag=%d)", tag);
}
break;
@@ -2201,11 +2228,16 @@ int toasty_process_events(ToastyFS *toasty, void **contexts, struct pollfd *poll
RequestQueue *reqs = NULL;
int tag = tcp_get_tag(&toasty->tcp, conn_idx);
CLIENT_TRACE("TCP EVENT: DISCONNECT (tag=%d)", tag);
if (tag == TAG_METADATA_SERVER) {
reqs = &toasty->metadata_server.reqs;
toasty->metadata_server.used = false;
CLIENT_TRACE("MS disconnect");
} else {
assert(tag > -1);
CLIENT_TRACE("CS disconnect");
if (toasty->chunk_servers[tag].connected)
reqs = &toasty->chunk_servers[tag].reqs;
@@ -2240,14 +2272,19 @@ int toasty_process_events(ToastyFS *toasty, void **contexts, struct pollfd *poll
case EVENT_MESSAGE:
{
CLIENT_TRACE("TCP EVENT: message");
for (;;) {
ByteView msg;
uint16_t msg_type;
int ret = tcp_next_message(&toasty->tcp, conn_idx, &msg, &msg_type);
if (ret == 0)
if (ret == 0) {
CLIENT_TRACE("Incomplete message");
break;
}
if (ret < 0) {
CLIENT_TRACE("Invalid message");
tcp_close(&toasty->tcp, conn_idx);
break;
}
@@ -2255,14 +2292,18 @@ int toasty_process_events(ToastyFS *toasty, void **contexts, struct pollfd *poll
RequestQueue *reqs;
int tag = tcp_get_tag(&toasty->tcp, conn_idx);
if (tag == TAG_METADATA_SERVER)
if (tag == TAG_METADATA_SERVER) {
CLIENT_TRACE("Message is from MS");
reqs = &toasty->metadata_server.reqs;
else
} else {
CLIENT_TRACE("Message is from CS");
reqs = &toasty->chunk_servers[tag].reqs;
}
Request req;
if (request_queue_pop(reqs, &req) < 0) {
// Unexpected message
CLIENT_TRACE("Message was unexpected");
tcp_consume_message(&toasty->tcp, conn_idx);
continue;
}
src/crash_logger.c
+433
View File
@@ -0,0 +1,433 @@
#if defined(__linux__) && defined(__x86_64__)
#define _GNU_SOURCE
#include <elf.h>
#include <fcntl.h>
#include <stdio.h>
#include <signal.h>
#include <unistd.h>
#include <string.h>
#include <assert.h>
#include <stdlib.h>
#include <stdbool.h>
#include <sys/reg.h>
#include <sys/stat.h>
#include <ucontext.h>
typedef struct {
uint64_t base_addr;
int count;
void *symbols;
void *strings;
} SymbolTable;
static int is_hex(char c)
{
return (c >= '0' && c <= '9') ||
(c >= 'a' && c <= 'f') ||
(c >= 'A' && c <= 'F');
}
static uint64_t query_base_addr(void)
{
int fd = open("/proc/self/maps", O_RDONLY);
if (fd < 0)
return -1;
char buf[128];
int ret = read(fd, buf, sizeof(buf));
if (ret < 0) {
close(fd);
return -1;
}
close(fd);
if (ret == 0 || !is_hex(buf[0]))
return -1;
int i = 0;
uint64_t base_addr = 0;
for (;;) {
char c = buf[i++];
int d;
if (0) {}
else if (c >= 'a' && c <= 'f') d = c - 'a' + 10;
else if (c >= 'A' && c <= 'F') d = c - 'A' + 10;
else d = c - '0';
if (base_addr > (UINT64_MAX - d) / 16)
return -1;
base_addr = base_addr * 16 + d;
if (i == ret)
return -1;
if (buf[i] == '-')
break;
if (!is_hex(buf[i]))
return -1;
}
return base_addr;
}
static int current_executable_path(char *dst, int cap)
{
if (cap == 0)
return -1;
int ret = readlink("/proc/self/exe", dst, cap-1);
if (ret < 0)
return -1;
dst[ret] = '\0';
return ret;
}
static int load_symbols_from_elf(void *src, int len, SymbolTable *st)
{
// NOTE: It's assumed is properly aligned
assert(((uintptr_t) src & 15) == 0);
// Check that the file contains a full header
if (len < (int) sizeof(Elf64_Ehdr))
return -1;
Elf64_Ehdr *ehdr = (Elf64_Ehdr*) src;
// Check that the file contains the full list
// of section headers
if (ehdr->e_shoff + ehdr->e_shnum * sizeof(Elf64_Shdr) > len)
return -1;
Elf64_Shdr *shdrs = (Elf64_Shdr*) (src + ehdr->e_shoff);
Elf64_Shdr *shstrtab_hdr = &shdrs[ehdr->e_shstrndx]; // TODO: bounds check
char *shstrtab = src + shstrtab_hdr->sh_offset;
// Iterate over the section headers to find the
// one reative to symbols and their strings
Elf64_Shdr *symtab_hdr = NULL;
Elf64_Shdr *strtab_hdr = NULL;
for (int i = 0; i < ehdr->e_shnum; i++) {
char *section_name = shstrtab + shdrs[i].sh_name;
if (0) {}
else if (!strcmp(section_name, ".symtab")) symtab_hdr = &shdrs[i];
else if (!strcmp(section_name, ".strtab")) strtab_hdr = &shdrs[i];
}
if (symtab_hdr == NULL || strtab_hdr == NULL) {
return -1;
}
void *mem = malloc(symtab_hdr->sh_size + strtab_hdr->sh_size);
if (mem == NULL) {
return -1;
}
st->count = symtab_hdr->sh_size / sizeof(Elf64_Sym);
st->symbols = mem;
st->strings = (char*) st->symbols + symtab_hdr->sh_size;
memcpy(st->symbols, src + symtab_hdr->sh_offset, symtab_hdr->sh_size);
memcpy(st->strings, src + strtab_hdr->sh_offset, strtab_hdr->sh_size);
return 0;
}
static char *read_file(char *path, int *len)
{
int fd = open(path, O_RDONLY);
if (fd < 0)
return NULL;
struct stat buf;
if (fstat(fd, &buf) < 0) {
close(fd);
return NULL;
}
*len = buf.st_size;
char *ptr = malloc(*len + 1);
if (ptr == NULL) {
close(fd);
return NULL;
}
for (int num = 0; num < *len; ) {
int ret = read(fd, ptr + num, *len - num);
if (ret <= 0) {
free(ptr);
close(fd);
return NULL;
}
num += ret;
}
ptr[*len] = '\0';
return ptr;
}
static int symbol_table_from_current_process(SymbolTable *st)
{
uint64_t base_addr = query_base_addr();
if (base_addr == (uint64_t) -1)
return -1;
st->base_addr = base_addr;
char path[1<<10];
if (current_executable_path(path, sizeof(path)) < 0)
return -1;
char *exe_ptr;
int exe_len;
exe_ptr = read_file(path, &exe_len);
if (exe_ptr == NULL)
return -1;
if (load_symbols_from_elf(exe_ptr, exe_len, st) < 0) {
free(exe_ptr);
return -1;
}
free(exe_ptr);
return 0;
}
static void symbol_table_free(SymbolTable *st)
{
free(st->symbols);
}
static char *symbol_table_find(SymbolTable *st, uint64_t addr)
{
for (int i = 0; i < st->count; i++) {
Elf64_Sym *sym = (Elf64_Sym*) st->symbols + i;
if (ELF64_ST_TYPE(sym->st_info) != STT_FUNC)
continue;
if (sym->st_value == 0)
continue;
uint64_t sym_beg = st->base_addr + sym->st_value;
uint64_t sym_end = st->base_addr + sym->st_value + sym->st_size;
if (addr >= sym_beg && addr < sym_end)
return (char*) st->strings + sym->st_name;
}
return NULL;
}
#if 0
static void symbol_table_dump(SymbolTable *st)
{
for (int i = 0; i < st->count; i++) {
Elf64_Sym *sym = (Elf64_Sym*) st->symbols + i;
if (ELF64_ST_TYPE(sym->st_info) != STT_FUNC)
continue;
if (sym->st_value == 0)
continue;
char *name = (char*) st->strings + sym->st_name;
printf("%s\n", name);
}
}
#endif
typedef struct {
char *name;
uint64_t addr;
} StackFrame;
static int walk_stack(uint64_t rip, uint64_t rbp, SymbolTable *st, StackFrame *frames, int max_frames)
{
int frame_count = 0;
if (frame_count < max_frames) {
frames[frame_count].addr = rip - st->base_addr;
frames[frame_count].name = symbol_table_find(st, rip);
frame_count++;
}
while (rbp != 0) {
if (rbp & 0xF)
break;
uint64_t *frame_ptr = (uint64_t*) rbp;
uint64_t next_rbp = frame_ptr[0];
uint64_t return_addr = frame_ptr[1];
if (next_rbp != 0 && next_rbp <= rbp)
break;
if (return_addr == 0)
break;
if (frame_count == max_frames)
break;
frames[frame_count].addr = return_addr - st->base_addr;
frames[frame_count].name = symbol_table_find(st, return_addr);
frame_count++;
rbp = next_rbp;
}
return frame_count;
}
static bool crash_logger_symbol_init = false;
static char* crash_logger_file_name = NULL;
static SymbolTable crash_logger_symbol_table;
static char* crash_logger_signal_stack;
static void crash_handler(int sig, siginfo_t *info, void *ucontext)
{
(void) info;
if (crash_logger_symbol_init) {
// Buffer for evaluating format strings
char tmp[1<<9];
int len;
ucontext_t *ctx = (ucontext_t*) ucontext;
uint64_t rip = ctx->uc_mcontext.gregs[REG_RIP];
uint64_t rbp = ctx->uc_mcontext.gregs[REG_RBP];
StackFrame frames[64];
int count = walk_stack(rip, rbp, &crash_logger_symbol_table, frames, 64);
int fd = open(crash_logger_file_name, O_WRONLY | O_CREAT, 0666);
if (fd < 0)
exit(1);
char *sig_name = "";
switch (sig) {
case SIGSEGV: sig_name = "Segmentation fault"; break;
case SIGBUS : sig_name = "Bus error"; break;
case SIGILL : sig_name = "Illegal instruction"; break;
case SIGFPE : sig_name = "Floating point exception"; break;
case SIGTRAP: sig_name = "Trace trap"; break;
case SIGSYS : sig_name = "Bad system call"; break;
case SIGABRT: sig_name = "Abort"; break;
}
if (sig_name[0] == '\0') {
len = snprintf(tmp, sizeof(tmp), "(unknown signal %d)\n", sig);
write(fd, tmp, len);
} else {
write(fd, sig_name, strlen(sig_name));
write(fd, "\n", 1);
}
for (int i = 0; i < count; i++) {
len = snprintf(tmp, sizeof(tmp), " [%d] 0x%lx %s\n", i, frames[i].addr,
frames[i].name ? frames[i].name : "?");
write(fd, tmp, len);
}
close(fd);
}
exit(1);
}
int crash_logger_init(char *file_name, int file_name_len)
{
{
char *file_name_copy = malloc(file_name_len + 1);
if (file_name_copy == NULL)
return -1;
memcpy(file_name_copy, file_name, file_name_len);
file_name_copy[file_name_len] = '\0';
crash_logger_file_name = file_name_copy;
}
if (symbol_table_from_current_process(&crash_logger_symbol_table) < 0) {
free(crash_logger_file_name);
return -1;
}
// Set up alternate signal stack
{
crash_logger_signal_stack = malloc(SIGSTKSZ);
if (crash_logger_signal_stack == NULL) {
symbol_table_free(&crash_logger_symbol_table);
free(crash_logger_file_name);
return -1;
}
stack_t ss;
ss.ss_sp = crash_logger_signal_stack;
ss.ss_size = SIGSTKSZ;
ss.ss_flags = 0;
if (sigaltstack(&ss, NULL) < 0) {
free(crash_logger_signal_stack);
symbol_table_free(&crash_logger_symbol_table);
free(crash_logger_file_name);
return -1;
}
}
{
// Register the crash handler
struct sigaction sa;
sa.sa_sigaction = crash_handler;
sa.sa_flags = SA_SIGINFO | SA_ONSTACK; // Add SA_ONSTACK flag
sigemptyset(&sa.sa_mask);
// Memory errors
sigaction(SIGSEGV, &sa, NULL); // Segmentation fault (invalid memory access)
sigaction(SIGBUS, &sa, NULL); // Bus error (misaligned access, hardware error)
// Execution errors
sigaction(SIGILL, &sa, NULL); // Illegal instruction
sigaction(SIGFPE, &sa, NULL); // Floating point exception
sigaction(SIGTRAP, &sa, NULL); // Trace trap
// System/resource errors
sigaction(SIGSYS, &sa, NULL); // Bad system call
sigaction(SIGABRT, &sa, NULL); // Abort (from assert, abort(), etc.)
// Optional: Resource limit violations
sigaction(SIGXCPU, &sa, NULL); // CPU time limit exceeded
sigaction(SIGXFSZ, &sa, NULL); // File size limit exceeded
}
crash_logger_symbol_init = true;
return 0;
}
void crash_logger_free(void)
{
if (!crash_logger_symbol_init)
return;
free(crash_logger_signal_stack);
symbol_table_free(&crash_logger_symbol_table);
free(crash_logger_file_name);
crash_logger_symbol_init = false;
}
#else
static int crash_logger_init(char *file_name, int file_name_len)
{
(void) file_name;
(void) file_name_len;
return -1;
}
static void crash_logger_free(void)
{
}
#endif
src/crash_logger.h
+7
View File
@@ -0,0 +1,7 @@
#ifndef CRASH_LOGGER_INCLUDED
#define CRASH_LOGGER_INCLUDED
int crash_logger_init(char *file_name, int file_name_len);
void crash_logger_free(void);
#endif // CRASH_LOGGER_INCLUDED
src/main_server.c
+63
View File
@@ -1,5 +1,6 @@
#ifdef BUILD_SERVER
#include <time.h>
#include <string.h>
#ifdef _WIN32
@@ -10,19 +11,49 @@
#define POLL poll
#endif
#include "crash_logger.h"
#include "chunk_server.h"
#include "metadata_server.h"
int metadata_server_main(int argc, char **argv)
{
{
time_t now = time(NULL);
struct tm unpacked_now;
#ifdef _WIN32
if (gmtime_s(&unpacked_now, &now) != 0)
return -1;
#else
if (gmtime_r(&now, &unpacked_now) == NULL)
return -1;
#endif
char time[sizeof("YYYYMMDDthhmmssz")];
int ret = strftime(time, sizeof(time),
"%Y%m%dT%H%M%SZ", &unpacked_now);
if (ret != sizeof(time)-1)
return -1;
char path[1<<10];
ret = snprintf(path, sizeof(path), "crash_%s_MS_%d.txt", time, getpid());
if (ret < 0 || ret >= (int) sizeof(path)) {
return -1;
}
int path_len = ret;
crash_logger_init(path, path_len);
}
void *contexts[TCP_POLL_CAPACITY];
struct pollfd polled[TCP_POLL_CAPACITY];
int num_polled;
int timeout = -1;
MetadataServer state;
num_polled = metadata_server_init(
&state, argc, argv, contexts, polled, &timeout);
if (num_polled < 0) return -1;
for (;;) {
POLL(polled, num_polled, timeout);
@@ -32,19 +63,49 @@ int metadata_server_main(int argc, char **argv)
if (num_polled < 0) return -1;
}
metadata_server_free(&state);
crash_logger_free();
return 0;
}
int chunk_server_main(int argc, char **argv)
{
{
time_t now = time(NULL);
struct tm unpacked_now;
#ifdef _WIN32
if (gmtime_s(&unpacked_now, &now) != 0)
return -1;
#else
if (gmtime_r(&now, &unpacked_now) == NULL)
return -1;
#endif
char time[sizeof("YYYYMMDDthhmmssz")];
int ret = strftime(time, sizeof(time),
"%Y%m%dT%H%M%SZ", &unpacked_now);
if (ret != sizeof(time)-1)
return -1;
char path[1<<10];
ret = snprintf(path, sizeof(path), "crash_%s_CS_%d.txt", time, getpid());
if (ret < 0 || ret >= (int) sizeof(path)) {
return -1;
}
int path_len = ret;
crash_logger_init(path, path_len);
}
void *contexts[TCP_POLL_CAPACITY];
struct pollfd polled[TCP_POLL_CAPACITY];
int num_polled;
int timeout = -1;
ChunkServer state;
num_polled = chunk_server_init(
&state, argc, argv, contexts, polled, &timeout);
if (num_polled < 0) return -1;
for (;;) {
POLL(polled, num_polled, timeout);
@@ -54,7 +115,9 @@ int chunk_server_main(int argc, char **argv)
&state, contexts, polled, num_polled, &timeout);
if (num_polled < 0) return -1;
}
chunk_server_free(&state);
crash_logger_free();
return 0;
}
src/metadata_server.c
+36 -1
View File
@@ -7,6 +7,8 @@
#include "message.h"
#include "metadata_server.h"
#define MS_TRACE(fmt, ...) fprintf(stderr, "MS: " fmt "\n", ##__VA_ARGS__);
static void chunk_server_peer_init(ChunkServerPeer *chunk_server, Time current_time)
{
chunk_server->used = true;
@@ -442,6 +444,28 @@ process_client_read(MetadataServer *state, int conn_idx, ByteView msg)
message_write(&writer, &len, sizeof(len));
message_write(&writer, desc.ptr, desc.len);
int locations[MAX_CHUNK_SERVERS];
int num_locations = choose_servers_for_write(state, locations, state->replication_factor);
assert(num_locations > -1 && num_locations < MAX_CHUNK_SERVERS);
if (num_locations > state->replication_factor)
num_locations = state->replication_factor;
uint32_t tmp_u32 = num_locations;
message_write(&writer, &tmp_u32, sizeof(tmp_u32));
for (int j = 0; j < num_locations; j++) {
int k = locations[j];
assert(k > -1);
assert(k < state->num_chunk_servers);
assert(state->chunk_servers[k].auth == true);
assert(state->chunk_servers[k].num_addrs > 0);
message_write_server_addr(&writer, &state->chunk_servers[k]);
}
if (!message_writer_free(&writer))
return -1;
@@ -1106,16 +1130,20 @@ int metadata_server_step(MetadataServer *state, void **contexts, struct pollfd *
switch (events[i].type) {
case EVENT_WAKEUP:
MS_TRACE("TCP EVENT: wakeup");
// Do nothing
break;
case EVENT_CONNECT:
MS_TRACE("TCP EVENT: connect");
tcp_set_tag(&state->tcp, conn_idx, CONNECTION_TAG_UNKNOWN, false);
break;
case EVENT_DISCONNECT:
{
MS_TRACE("TCP EVENT: disconnect");
if (events[i].tag >= 0) {
MS_TRACE("Chunk server disconnected");
chunk_server_peer_free(&state->chunk_servers[events[i].tag]);
assert(state->num_chunk_servers > 0);
state->num_chunk_servers--;
@@ -1125,14 +1153,19 @@ int metadata_server_step(MetadataServer *state, void **contexts, struct pollfd *
case EVENT_MESSAGE:
{
// We don't trace message events from chunk servers
// as it would become very verbose
for (;;) {
ByteView msg;
uint16_t msg_type;
int ret = tcp_next_message(&state->tcp, conn_idx, &msg, &msg_type);
if (ret == 0)
if (ret == 0) {
MS_TRACE("Incomplete message");
break;
}
if (ret < 0) {
MS_TRACE("Invalid message");
tcp_close(&state->tcp, conn_idx);
break;
}
@@ -1167,12 +1200,14 @@ int metadata_server_step(MetadataServer *state, void **contexts, struct pollfd *
int tag = tcp_get_tag(&state->tcp, conn_idx);
if (tag == CONNECTION_TAG_CLIENT) {
MS_TRACE("Message from client");
ret = process_client_message(state, conn_idx, msg_type, msg);
} else {
state->chunk_servers[tag].last_response_time = current_time;
ret = process_chunk_server_message(state, conn_idx, msg_type, msg);
}
if (ret < 0) {
MS_TRACE("Message processing failure");
tcp_close(&state->tcp, conn_idx);
break;
}