redis_server_protocol: fix mem leak

src/brpc/policy/redis_protocol.cpp

@@ -80,6 +80,9 @@ public:
 class RedisConnContext : public brpc::SharedObject {
 public:
     RedisConnContext() : handler_continue(NULL) {}
+    ~RedisConnContext() {
+        ClearQueue(dones);
+    }
 
     void Push(ConsumeTaskDone* done) {
         std::unique_lock<butil::Mutex> m(_mutex);
@@ -120,9 +123,13 @@ public:
                 ready_to_delete.push(head);
             }
         }
-        while (!ready_to_delete.empty()) {
-            ConsumeTaskDone* head = ready_to_delete.front();
-            ready_to_delete.pop();
+        ClearQueue(ready_to_delete);
+    }
+
+    void ClearQueue(std::queue<ConsumeTaskDone*>& queue) {
+        while (!queue.empty()) {
+            ConsumeTaskDone* head = queue.front();
+            queue.pop();
             delete head;
         }
     }

src/brpc/redis_message.cpp

@@ -375,8 +375,15 @@ void RedisMessage::CopyFromDifferentArena(const RedisMessage& other,
             new (&subs[i]) RedisMessage;
         }
         _data.array.last_index = other._data.array.last_index;
-        for (size_t i = 0; i < _length; ++i) {
-            subs[i].CopyFromDifferentArena(other._data.array.replies[i], arena);
+        if (_data.array.last_index > 0) {
+            // incomplete state
+            for (int i = 0; i < _data.array.last_index; ++i) {
+                subs[i].CopyFromDifferentArena(other._data.array.replies[i], arena);
+            }
+        } else {
+            for (size_t i = 0; i < _length; ++i) {
+                subs[i].CopyFromDifferentArena(other._data.array.replies[i], arena);
+            }
         }
         _data.array.replies = subs;
     }