fix(duplication): fix duplication core because of removing or pausing

[lengyuexuexuan]

What problem does this PR solve?

#2211

What is changed and how does it work?

1. Background
   During duplicationof a table, if the commands dup remove/pause are executed or a balance operation is performed at the same time, there is a chance that a node may core dump with signal ID 11. The core dump locations vary, but they all have one thing in common: they occur during memory allocation or deallocation.

2. Analysis
   Based on extensive testing, the following conclusions can be drawn:
   a. The issue only reproduces when there is write traffic. The difference between having and not having write traffic is: It adds the ship and load_private_log tasks.
   b. The core dump occurs during the execution of cancel_all().
   c. The issue occurs with low probability (approximately 1 in 100).

   Through analysis using ASAN (AddressSanitizer):
   dup_remove_asan.txt
   Based on ASAN analysis, the following conclusions can be drawn:
   a. The memory corruption occurs during the ship process. The mutations obtained from replaying the plog are passed to ship, leading to the issue.
   b. _load_mutations is captured by a lambda expression and then passed to a std::function. Since std::move is used, the lifetime of _load_mutations is tied to that of the std::function.
   c. The cancel_all() function is executed in the default thread pool. At this point, the following function is called. When the std::function is set to nullptr, it will release the memory it manages.

   incubator-pegasus/src/task/task.h

   Line 341 in e64faa7

   void clear_non_trivial_on_task_end() override { _cb = nullptr; }

   
   d. However, each task executes exec_internal() in its own thread pool, and eventually calls release_ref(), which results in delete this.

   incubator-pegasus/src/task/task.cpp

   Line 224 in e64faa7

   this->release_ref(); // added in enqueue(pool)

3. Conclusion
   Both task.cancel() and task.exec_internal() destruct the std::function member inside the task object. These two operations are executed in different threads, and there is no mechanism in place to prevent race conditions between them. As a result, it is possible for both threads to attempt to destruct the same std::function, which can lead to a double deletion of the memory associated with _load_mutations. This ultimately causes memory corruption.

4. Solution
   Replace cancel_all() with wait_all().

Tests

• Manual test
  Based on 20 test runs, the issue was successfully avoided, confirming the effectiveness of the solution.