Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion src/networking.c
Original file line number Diff line number Diff line change
Expand Up @@ -283,7 +283,9 @@ static int isCopyAvoidPreferred(client *c, robj *obj) {
}

client *createClient(connection *conn) {
client *c = zmalloc(sizeof(client));
/* client carries _Alignas(CACHE_LINE_SIZE) members; plain zmalloc only
* guarantees 16-byte alignment. */
client *c = zmalloc_cache_aligned(sizeof(client));

/* passing NULL as conn it is possible to create a non connected client.
* This is useful since all the commands needs to be executed
Expand Down
29 changes: 18 additions & 11 deletions src/server.h
Original file line number Diff line number Diff line change
Expand Up @@ -1334,7 +1334,6 @@ typedef struct client {
list *reply; /* List of reply objects to send to the client. */
listNode *io_last_reply_block; /* Last client reply block when sent to IO thread */
size_t io_last_bufpos; /* The client's bufpos at the time it was sent to the IO thread */
LastWrittenBuf io_last_written; /* Track state for last written buffer */
unsigned long long reply_bytes; /* Tot bytes of objects in reply list. */
listNode clients_pending_write_node; /* list node in clients_pending_write or in clients_pending_io_write list */
size_t bufpos;
Expand All @@ -1353,28 +1352,21 @@ typedef struct client {
/* Cache Locality: Grouped with 'flag' for getClientType() hot path. */
slotMigrationJob *slot_migration_job; /* Pointer to the slot migration job, or NULL. */
uint16_t write_flags; /* Client Write flags - used to communicate the client write state. */
volatile uint8_t io_read_state; /* Indicate the IO read state of the client */
volatile uint8_t io_write_state; /* Indicate the IO write state of the client */
uint8_t resp; /* RESP protocol version. Can be 2 or 3. */
uint8_t cur_tid; /* ID of IO thread currently performing IO for this client */
/* In updateClientMemoryUsage() we track the memory usage of
* each client and add it to the sum of all the clients of a given type,
* however we need to remember what was the old contribution of each
* client, and in which category the client was, in order to remove it
* before adding it the new value. */
uint8_t last_memory_type;
uint8_t capa; /* Client capabilities: CLIENT_CAPA* macros. */
/* Statistics and metrics */
/* Statistics and metrics (main-thread writers; see the I/O thread block
* below for counters the I/O threads update). */
unsigned long long net_input_bytes; /* Total network input bytes read from this client. */
unsigned long long net_input_bytes_curr_cmd; /* Total network input bytes read for the* execution of this client's current command. */
unsigned long long net_output_bytes; /* Total network output bytes sent to this client. */
unsigned long long commands_processed; /* Total count of commands this client executed. */
unsigned long long net_output_bytes_curr_cmd; /* Total network output bytes sent to this client, by the current command. */
_Atomic(size_t) io_tracked_reply_len; /* Total size of BULK_STR_REF replies tracked by I/O threads. */
size_t buf_peak; /* Peak used size of buffer in last 5 sec interval. */
int nwritten; /* Number of bytes of the last write. */
int nread; /* Number of bytes of the last read. */
int read_flags; /* Client Read flags - used to communicate the client read state. */
int slot; /* The slot the client is executing against. Set to -1 if no slot is being used */
listNode *mem_usage_bucket_node;
clientMemUsageBucket *mem_usage_bucket;
Expand All @@ -1384,8 +1376,23 @@ typedef struct client {
* client, and in which category the client was, in order to remove it
* before adding it the new value. */
size_t last_memory_usage;
/* Fields written by the owning I/O thread during offloaded jobs. The main
* thread may append replies and update its own counters concurrently with
* an in-flight job (see writevToClient), so these live on a dedicated
* cache line: sharing one with any main-thread-written field makes every
* I/O syscall and every reply chunk trade that line between the two
* threads. Requires client allocations to be CACHE_LINE_SIZE-aligned. */
_Alignas(CACHE_LINE_SIZE) volatile uint8_t io_read_state; /* Indicate the IO read state of the client */
volatile uint8_t io_write_state; /* Indicate the IO write state of the client */
uint8_t cur_tid; /* ID of IO thread currently performing IO for this client */
int read_flags; /* Client Read flags - used to communicate the client read state. */
int nread; /* Number of bytes of the last read. */
int nwritten; /* Number of bytes of the last write. */
_Atomic(size_t) io_tracked_reply_len; /* Total size of BULK_STR_REF replies tracked by I/O threads. */
unsigned long long net_input_bytes_curr_cmd; /* Total network input bytes read for the execution of this client's current command. */
LastWrittenBuf io_last_written; /* Track state for last written buffer */
/* Fields after this point are less frequently used */
listNode *client_list_node; /* list node in client list */
_Alignas(CACHE_LINE_SIZE) listNode *client_list_node; /* list node in client list */
Comment thread
coderabbitai[bot] marked this conversation as resolved.
mstime_t buf_peak_last_reset_time; /* keeps the last time the buffer peak value was reset */
size_t querybuf_peak; /* Recent (100ms or more) peak of querybuf size. */
dictEntry *cur_script; /* Cached pointer to the dictEntry of the script being executed. */
Expand Down
18 changes: 13 additions & 5 deletions src/unit/test_networking.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -137,6 +137,14 @@ static fakeConnection *connCreateFake(void) {
return conn;
}

/* client carries _Alignas(CACHE_LINE_SIZE) fields, so plain zmalloc/zcalloc
* would under-align it. */
static client *allocTestClient(void) {
client *c = (client *)(zmalloc_cache_aligned(sizeof(client)));
memset(c, 0, sizeof(client));
return c;
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.

/* Test fixture for networking tests - minimal fixture with no setup/teardown */
class NetworkingTest : public ::testing::Test {
protected:
Expand All @@ -156,7 +164,7 @@ class NetworkingTest : public ::testing::Test {
};

TEST_F(NetworkingTest, TestWriteToReplica) {
client *c = (client *)(zcalloc(sizeof(client)));
client *c = allocTestClient();
initClientReplicationData(c);
server.repl_buffer_blocks = listCreate();
/* Ensure replicas list exists before creating backlog */
Expand Down Expand Up @@ -296,7 +304,7 @@ TEST_F(NetworkingTest, TestWriteToReplica) {
}

TEST_F(NetworkingTest, TestPostWriteToReplica) {
client *c = (client *)(zcalloc(sizeof(client)));
client *c = allocTestClient();
initClientReplicationData(c);
server.repl_buffer_blocks = listCreate();
/* Ensure replicas list exists before creating backlog */
Expand Down Expand Up @@ -412,7 +420,7 @@ TEST_F(NetworkingTest, TestPostWriteToReplica) {
}

TEST_F(NetworkingTest, TestBackupAndUpdateClientArgv) {
client *c = (client *)(zmalloc(sizeof(client)));
client *c = allocTestClient();
/* Test 1: Initial backup of arguments */
c->argc = 2;
robj **initial_argv = (robj **)(zmalloc(sizeof(robj *) * 2));
Expand Down Expand Up @@ -473,7 +481,7 @@ TEST_F(NetworkingTest, TestBackupAndUpdateClientArgv) {
}

TEST_F(NetworkingTest, TestRewriteClientCommandArgument) {
client *c = (client *)(zmalloc(sizeof(client)));
client *c = allocTestClient();
c->argc = 3;
robj **initial_argv = (robj **)(zmalloc(sizeof(robj *) * 3));
c->argv = initial_argv;
Expand Down Expand Up @@ -526,7 +534,7 @@ TEST_F(NetworkingTest, TestRewriteClientCommandArgument) {

/* Helper function to create test client */
static client *createTestClient(void) {
client *c = (client *)(zcalloc(sizeof(client)));
client *c = allocTestClient();

c->buf = (char *)zmalloc_usable(PROTO_REPLY_CHUNK_BYTES, &c->buf_usable_size);
c->reply = listCreate();
Expand Down
Loading