Add slowlog entry for slowest sub command

src/command.c

@@ -280,6 +280,9 @@ int cmd_forward_multikey(struct command *cmd, struct redis_data *data, const cha
         memcpy(&ncmd->req_buf[0], &key->buf[0], sizeof(key->buf[0]));
         memcpy(&ncmd->req_buf[1], &key->buf[1], sizeof(key->buf[1]));
         ncmd->prefix = (char*)prefix;
+        ncmd->data.type = REP_ARRAY;
+        ncmd->data.elements = 1;
+        ncmd->data.element = key;
 
         if (cmd_forward_basic(ncmd) == CORVUS_ERR) {
             cmd_mark_fail(ncmd, rep_forward_err);
@@ -321,6 +324,9 @@ int cmd_forward_mset(struct command *cmd, struct redis_data *data)
         memcpy(&ncmd->req_buf[0], &key->buf[0], sizeof(key->buf[0]));
         memcpy(&ncmd->req_buf[1], &value->buf[1], sizeof(value->buf[1]));
         ncmd->prefix = (char*)rep_set;
+        ncmd->data.type = REP_ARRAY;
+        ncmd->data.elements = 2;
+        ncmd->data.element = &data->element[i];
 
         if (cmd_forward_basic(ncmd) == CORVUS_ERR) {
             cmd_mark_fail(ncmd, rep_forward_err);
@@ -1138,6 +1144,10 @@ void cmd_stats(struct command *cmd, int64_t end_time)
             struct slowlog_entry *entry = slowlog_create_entry(cmd,
                 remote_latency / 1000, total_latency / 1000);
             slowlog_set(&cmd->ctx->slowlog, entry);
+            entry = slowlog_create_sub_entry(cmd, total_latency / 1000);
+            if (entry) {
+                slowlog_set(&cmd->ctx->slowlog, entry);
+            }
         }
     }
 
@@ -1224,7 +1234,9 @@ void cmd_free(struct command *cmd)
     struct command *c;
     struct context *ctx = cmd->ctx;
 
-    if (cmd->data.type != REP_UNKNOWN) {
+    // When cmd->prefix is NULL it's a sub command,
+    // cmd->data of sub command should never be deallocated.
+    if (cmd->data.type != REP_UNKNOWN && cmd->prefix == NULL) {
         redis_data_free(&cmd->data);
         cmd->data.type = REP_UNKNOWN;
     }

src/command.h

@@ -232,7 +232,15 @@ struct command {
     bool stale;
     bool cmd_fail;
 
-    struct redis_data data;  /* for slowlog */
+    /* For slowlog
+       When used in parent cmd or non-multiple-key command,
+       it contains all command data. When used in sub command,
+       it does not contain command name because it's already
+       in command.prefix. Note that in sub command it only contains
+       weak reference to command.data of parent cmd and should never
+       be deallocated.
+    */
+    struct redis_data data;
 };
 
 struct redirect_info {

src/slowlog.c

@@ -111,6 +111,30 @@ struct slowlog_entry *slowlog_create_entry(struct command *cmd, int64_t remote_l
     return entry;
 }
 
+// Return NULL when `cmd` is not a multiple keys command
+struct slowlog_entry *slowlog_create_sub_entry(struct command *cmd, int64_t total_latency)
+{
+    switch (cmd->cmd_type) {
+        case CMD_MGET:
+        case CMD_MSET:
+        case CMD_DEL:
+        case CMD_EXISTS:
+            break;
+        default:
+            return NULL;
+    }
+    int64_t max_remote_latency = 0;
+    struct command *c, *slowest_sub_cmd = NULL;
+    STAILQ_FOREACH(c, &cmd->sub_cmds, sub_cmd_next) {
+        int64_t remote_latency = cmd->rep_time[1] - cmd->rep_time[0];
+        if (remote_latency > max_remote_latency) {
+            max_remote_latency = remote_latency;
+            slowest_sub_cmd = c;
+        }
+    }
+    return slowlog_create_entry(slowest_sub_cmd, max_remote_latency / 1000, total_latency);
+}
+
 void slowlog_dec_ref(struct slowlog_entry *entry)
 {
     int refcount = ATOMIC_DEC(entry->refcount, 1);

src/slowlog.h

@@ -39,6 +39,8 @@ void slowlog_free(struct slowlog_queue *slowlog);
 
 // only called by the thread who creates the log
 struct slowlog_entry *slowlog_create_entry(struct command *cmd, int64_t remote_latency, int64_t total_latency);
+// create entry for the slowlest sub cmd, its total_latency should be the same as parent cmd
+struct slowlog_entry *slowlog_create_sub_entry(struct command *cmd, int64_t total_latency);
 // called by all worker threads
 void slowlog_set(struct slowlog_queue *queue, struct slowlog_entry *entry);
 void slowlog_dec_ref(struct slowlog_entry *entry);