diff --git a/python/ray/includes/ray_config.pxd b/python/ray/includes/ray_config.pxd index 5316e5c00..bf48176f8 100644 --- a/python/ray/includes/ray_config.pxd +++ b/python/ray/includes/ray_config.pxd @@ -49,8 +49,6 @@ cdef extern from "ray/common/ray_config.h" nogil: int object_manager_push_timeout_ms() const - int object_manager_repeated_push_delay_ms() const - uint64_t object_manager_default_chunk_size() const int num_workers_per_process_python() const diff --git a/python/ray/includes/ray_config.pxi b/python/ray/includes/ray_config.pxi index 0d910bccf..e50af1a3c 100644 --- a/python/ray/includes/ray_config.pxi +++ b/python/ray/includes/ray_config.pxi @@ -84,10 +84,6 @@ cdef class Config: def object_manager_push_timeout_ms(): return RayConfig.instance().object_manager_push_timeout_ms() - @staticmethod - def object_manager_repeated_push_delay_ms(): - return RayConfig.instance().object_manager_repeated_push_delay_ms() - @staticmethod def object_manager_default_chunk_size(): return RayConfig.instance().object_manager_default_chunk_size() diff --git a/python/ray/tests/test_multinode_failures_2.py b/python/ray/tests/test_multinode_failures_2.py index be855306c..3dc65be55 100644 --- a/python/ray/tests/test_multinode_failures_2.py +++ b/python/ray/tests/test_multinode_failures_2.py @@ -22,7 +22,6 @@ import ray.ray_constants as ray_constants "num_heartbeats_timeout": 10, "object_manager_pull_timeout_ms": 1000, "object_manager_push_timeout_ms": 1000, - "object_manager_repeated_push_delay_ms": 1000, }, }], indirect=True) diff --git a/python/ray/tests/test_object_manager.py b/python/ray/tests/test_object_manager.py index b7252519b..57074df1a 100644 --- a/python/ray/tests/test_object_manager.py +++ b/python/ray/tests/test_object_manager.py @@ -199,15 +199,11 @@ def test_actor_broadcast(ray_start_cluster_with_resource): def test_object_transfer_retry(ray_start_cluster): cluster = ray_start_cluster - repeated_push_delay = 1 - # Force the sending object manager to allow duplicate pushes again sooner. # Also, force the receiving object manager to retry the pull sooner. We # make the chunk size smaller in order to make it easier to test objects # with multiple chunks. config = { - "object_manager_repeated_push_delay_ms": repeated_push_delay * 1000, - "object_manager_pull_timeout_ms": repeated_push_delay * 1000 / 4, "object_manager_default_chunk_size": 1000, "object_store_full_max_retries": 1, } @@ -230,13 +226,7 @@ def test_object_transfer_retry(ray_start_cluster): # Get the objects locally to cause them to be transferred. This is the # first time the objects are getting transferred, so it should happen # quickly. - start_time = time.time() ray.get(x_id) - end_time = time.time() - if end_time - start_time > repeated_push_delay: - warnings.warn("The initial transfer took longer than the repeated " - "push delay, so this test may not be testing the thing " - "it's supposed to test.") def not_exists(): return not ray.worker.global_worker.core_worker.object_exists(x_id) @@ -257,20 +247,6 @@ def test_object_transfer_retry(ray_start_cluster): # Get the object again and make sure it gets transferred. ray.get(x_id) - end_transfer_time = time.time() - # We should have had to wait for the repeated push delay. - assert end_transfer_time - start_time >= repeated_push_delay - - # Force the object to be evicted again and wait longer than the repeated - # push delay and make sure that the object is transferred again. - force_eviction() - time.sleep(repeated_push_delay) - - # Fetch the object again. This should not wait for the delay. - start_time = time.time() - ray.get(x_id) - end_time = time.time() - assert end_time - start_time < repeated_push_delay # The purpose of this test is to make sure we can transfer many objects. In the diff --git a/src/ray/common/ray_config_def.h b/src/ray/common/ray_config_def.h index f8c109de7..b9b33df36 100644 --- a/src/ray/common/ray_config_def.h +++ b/src/ray/common/ray_config_def.h @@ -185,10 +185,6 @@ RAY_CONFIG(int, object_manager_pull_timeout_ms, 10000) /// 0: giving up retrying immediately. RAY_CONFIG(int, object_manager_push_timeout_ms, 10000) -/// The period of time that an object manager will wait before pushing the -/// same object again to a specific object manager. -RAY_CONFIG(int, object_manager_repeated_push_delay_ms, 60000) - /// Default chunk size for multi-chunk transfers to use in the object manager. /// In the object manager, no single thread is permitted to transfer more /// data than what is specified by the chunk size unless the number of object diff --git a/src/ray/object_manager/object_manager.cc b/src/ray/object_manager/object_manager.cc index 64d8ca94f..d5dbd7721 100644 --- a/src/ray/object_manager/object_manager.cc +++ b/src/ray/object_manager/object_manager.cc @@ -404,56 +404,6 @@ void ObjectManager::HandleReceiveFinished(const ObjectID &object_id, profile_events_.push_back(profile_event); } -void PushManager::StartPush(const UniqueID &push_id, int64_t num_chunks, - std::function send_chunk_fn) { - RAY_LOG(DEBUG) << "Start push for " << push_id << ", num chunks " << num_chunks; - RAY_CHECK(num_chunks > 0); - push_info_[push_id] = std::make_pair(num_chunks, send_chunk_fn); - next_chunk_id_[push_id] = 0; - chunks_remaining_ += num_chunks; - ScheduleRemainingPushes(); - RAY_CHECK(push_info_.size() == next_chunk_id_.size()); -} - -void PushManager::OnChunkComplete() { - chunks_in_flight_ -= 1; - ScheduleRemainingPushes(); -} - -void PushManager::ScheduleRemainingPushes() { - // Loop over all active pushes for approximate round-robin prioritization. - // TODO(ekl) this isn't the best implementation of round robin, we should - // consider tracking the number of chunks active per-push and balancing those. - while (chunks_in_flight_ < max_chunks_in_flight_ && push_info_.size() > 0) { - // Loop over each active push and try to send another chunk. - auto it = push_info_.begin(); - while (it != push_info_.end() && chunks_in_flight_ < max_chunks_in_flight_) { - auto push_id = it->first; - auto max_chunks = it->second.first; - auto send_chunk_fn = it->second.second; - - // Send the next chunk for this push. - send_chunk_fn(next_chunk_id_[push_id]); - chunks_in_flight_ += 1; - chunks_remaining_ -= 1; - RAY_LOG(DEBUG) << "Sending chunk " << next_chunk_id_[push_id] << " of " - << max_chunks << " for push " << push_id << ", chunks in flight " - << NumChunksInFlight() << " / " << max_chunks_in_flight_ - << " max, remaining chunks: " << NumChunksRemaining(); - - // It is the last chunk and we don't need to track it any more. - if (++next_chunk_id_[push_id] >= max_chunks) { - next_chunk_id_.erase(push_id); - push_info_.erase(it++); - RAY_LOG(DEBUG) << "Push for " << push_id - << " completed, remaining: " << NumPushesInFlight(); - } else { - it++; - } - } - } -} - void ObjectManager::Push(const ObjectID &object_id, const NodeID &client_id) { RAY_LOG(DEBUG) << "Push on " << self_node_id_ << " to " << client_id << " of object " << object_id; @@ -489,29 +439,6 @@ void ObjectManager::Push(const ObjectID &object_id, const NodeID &client_id) { return; } - // If we haven't pushed this object to this same object manager yet, then push - // it. If we have, but it was a long time ago, then push it. If we have and it - // was recent, then don't do it again. - auto &recent_pushes = local_objects_[object_id].recent_pushes; - auto it = recent_pushes.find(client_id); - if (it == recent_pushes.end()) { - // We haven't pushed this specific object to this specific object manager - // yet (or if we have then the object must have been evicted and recreated - // locally). - recent_pushes[client_id] = absl::GetCurrentTimeNanos() / 1000000; - } else { - int64_t current_time = absl::GetCurrentTimeNanos() / 1000000; - if (current_time - it->second <= - RayConfig::instance().object_manager_repeated_push_delay_ms()) { - // We pushed this object to the object manager recently, so don't do it - // again. - RAY_LOG(DEBUG) << "Object " << object_id << " recently pushed to " << client_id; - return; - } else { - it->second = current_time; - } - } - auto rpc_client = GetRpcClient(client_id); if (rpc_client) { const object_manager::protocol::ObjectInfoT &object_info = @@ -532,10 +459,11 @@ void ObjectManager::Push(const ObjectID &object_id, const NodeID &client_id) { << ", total data size: " << data_size; UniqueID push_id = UniqueID::FromRandom(); - push_manager_->StartPush(push_id, num_chunks, [=](int64_t chunk_id) { + push_manager_->StartPush(client_id, object_id, num_chunks, [=](int64_t chunk_id) { SendObjectChunk(push_id, object_id, owner_address, client_id, data_size, - metadata_size, chunk_id, rpc_client, - [=](const Status &status) { push_manager_->OnChunkComplete(); }); + metadata_size, chunk_id, rpc_client, [=](const Status &status) { + push_manager_->OnChunkComplete(client_id, object_id); + }); }); } else { // Push is best effort, so do nothing here. diff --git a/src/ray/object_manager/object_manager.h b/src/ray/object_manager/object_manager.h index cba152b58..0ab6ef049 100644 --- a/src/ray/object_manager/object_manager.h +++ b/src/ray/object_manager/object_manager.h @@ -38,6 +38,7 @@ #include "ray/object_manager/object_directory.h" #include "ray/object_manager/ownership_based_object_directory.h" #include "ray/object_manager/plasma/store_runner.h" +#include "ray/object_manager/push_manager.h" #include "ray/rpc/object_manager/object_manager_client.h" #include "ray/rpc/object_manager/object_manager_server.h" @@ -76,9 +77,6 @@ struct ObjectManagerConfig { struct LocalObjectInfo { /// Information from the object store about the object. object_manager::protocol::ObjectInfoT object_info; - /// A map from the ID of a remote object manager to the timestamp of when - /// the object was last pushed to that object manager (if a push took place). - std::unordered_map recent_pushes; }; class ObjectStoreRunner { @@ -99,75 +97,6 @@ class ObjectManagerInterface { virtual ~ObjectManagerInterface(){}; }; -class PushManager { - public: - /// Manages rate limiting of outbound object pushes. - - /// Create a push manager. - /// - /// \param max_chunks_in_flight Max number of chunks allowed to be in flight - /// from this PushManager (this raylet). - PushManager(int64_t max_chunks_in_flight) - : max_chunks_in_flight_(max_chunks_in_flight) { - RAY_CHECK(max_chunks_in_flight_ > 0) << max_chunks_in_flight_; - }; - - /// Start pushing an object subject to max chunks in flight limit. - /// - /// \param push_id Unique identifier for this push. - /// \param num_chunks The total number of chunks to send. - /// \param send_chunk_fn This function will be called with args 0...{num_chunks-1}. - /// The caller promises to call PushManager::OnChunkComplete() - /// once a call to send_chunk_fn finishes. - void StartPush(const UniqueID &push_id, int64_t num_chunks, - std::function send_chunk_fn); - - /// Called every time a chunk completes to trigger additional sends. - /// TODO(ekl) maybe we should cancel the entire push on error. - void OnChunkComplete(); - - /// Return the number of chunks currently in flight. For testing only. - int64_t NumChunksInFlight() const { return chunks_in_flight_; }; - - /// Return the number of chunks remaining. For testing only. - int64_t NumChunksRemaining() const { return chunks_remaining_; }; - - /// Return the number of pushes currently in flight. For testing only. - int64_t NumPushesInFlight() const { return push_info_.size(); }; - - std::string DebugString() const { - std::stringstream result; - result << "PushManager:"; - result << "\n- num pushes in flight: " << NumPushesInFlight(); - result << "\n- num chunks in flight: " << NumChunksInFlight(); - result << "\n- num chunks remaining: " << NumChunksRemaining(); - result << "\n- max chunks allowed: " << max_chunks_in_flight_; - return result.str(); - } - - private: - /// Called on completion events to trigger additional pushes. - void ScheduleRemainingPushes(); - - /// Info about the pushed object: (num_chunks total, chunk_send_fn). - typedef std::pair> PushInfo; - - /// Max number of chunks in flight allowed. - const int64_t max_chunks_in_flight_; - - /// Running count of chunks remaining to send. - int64_t chunks_remaining_ = 0; - - /// Running count of chunks in flight, used to limit progress of in_flight_pushes_. - int64_t chunks_in_flight_ = 0; - - /// Tracks all pushes with chunk transfers in flight. - absl::flat_hash_map push_info_; - - /// Tracks progress of in flight pushes. - absl::flat_hash_map next_chunk_id_; -}; - // TODO(hme): Add success/failure callbacks for push and pull. class ObjectManager : public ObjectManagerInterface, public rpc::ObjectManagerServiceHandler { diff --git a/src/ray/object_manager/push_manager.cc b/src/ray/object_manager/push_manager.cc new file mode 100644 index 000000000..531980833 --- /dev/null +++ b/src/ray/object_manager/push_manager.cc @@ -0,0 +1,75 @@ +// Copyright 2017 The Ray Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#include "ray/object_manager/push_manager.h" + +#include "ray/common/common_protocol.h" +#include "ray/util/util.h" + +namespace ray { + +void PushManager::StartPush(const NodeID &dest_id, const ObjectID &obj_id, + int64_t num_chunks, + std::function send_chunk_fn) { + auto push_id = std::make_pair(dest_id, obj_id); + if (push_info_.contains(push_id)) { + RAY_LOG(DEBUG) << "Duplicate push request " << push_id.first << ", " + << push_id.second; + return; + } + RAY_CHECK(num_chunks > 0); + push_info_[push_id].reset(new PushState(num_chunks, send_chunk_fn)); + ScheduleRemainingPushes(); +} + +void PushManager::OnChunkComplete(const NodeID &dest_id, const ObjectID &obj_id) { + auto push_id = std::make_pair(dest_id, obj_id); + chunks_in_flight_ -= 1; + if (--push_info_[push_id]->chunks_remaining <= 0) { + push_info_.erase(push_id); + RAY_LOG(DEBUG) << "Push for " << push_id.first << ", " << push_id.second + << " completed, remaining: " << NumPushesInFlight(); + } + ScheduleRemainingPushes(); +} + +void PushManager::ScheduleRemainingPushes() { + bool keep_looping = true; + // Loop over all active pushes for approximate round-robin prioritization. + // TODO(ekl) this isn't the best implementation of round robin, we should + // consider tracking the number of chunks active per-push and balancing those. + while (chunks_in_flight_ < max_chunks_in_flight_ && keep_looping) { + // Loop over each active push and try to send another chunk. + auto it = push_info_.begin(); + keep_looping = false; + while (it != push_info_.end() && chunks_in_flight_ < max_chunks_in_flight_) { + auto push_id = it->first; + auto &info = it->second; + if (info->next_chunk_id < info->num_chunks) { + // Send the next chunk for this push. + info->chunk_send_fn(info->next_chunk_id++); + chunks_in_flight_ += 1; + keep_looping = true; + RAY_LOG(DEBUG) << "Sending chunk " << info->next_chunk_id << " of " + << info->num_chunks << " for push " << push_id.first << ", " + << push_id.second << ", chunks in flight " << NumChunksInFlight() + << " / " << max_chunks_in_flight_ + << " max, remaining chunks: " << NumChunksRemaining(); + } + it++; + } + } +} + +} // namespace ray diff --git a/src/ray/object_manager/push_manager.h b/src/ray/object_manager/push_manager.h new file mode 100644 index 000000000..9e2077fa6 --- /dev/null +++ b/src/ray/object_manager/push_manager.h @@ -0,0 +1,118 @@ +// Copyright 2017 The Ray Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#pragma once + +#include +#include + +#include "absl/container/flat_hash_map.h" +#include "absl/container/flat_hash_set.h" +#include "ray/common/id.h" +#include "ray/common/ray_config.h" +#include "ray/common/status.h" + +namespace ray { + +/// Manages rate limiting and deduplication of outbound object pushes. +class PushManager { + public: + /// Create a push manager. + /// + /// \param max_chunks_in_flight Max number of chunks allowed to be in flight + /// from this PushManager (this raylet). + PushManager(int64_t max_chunks_in_flight) + : max_chunks_in_flight_(max_chunks_in_flight) { + RAY_CHECK(max_chunks_in_flight_ > 0) << max_chunks_in_flight_; + }; + + /// Start pushing an object subject to max chunks in flight limit. + /// + /// Duplicate concurrent pushes to the same destination will be suppressed. + /// + /// \param dest_id The node to send to. + /// \param obj_id The object to send. + /// \param num_chunks The total number of chunks to send. + /// \param send_chunk_fn This function will be called with args 0...{num_chunks-1}. + /// The caller promises to call PushManager::OnChunkComplete() + /// once a call to send_chunk_fn finishes. + void StartPush(const NodeID &dest_id, const ObjectID &obj_id, int64_t num_chunks, + std::function send_chunk_fn); + + /// Called every time a chunk completes to trigger additional sends. + /// TODO(ekl) maybe we should cancel the entire push on error. + void OnChunkComplete(const NodeID &dest_id, const ObjectID &obj_id); + + /// Return the number of chunks currently in flight. For testing only. + int64_t NumChunksInFlight() const { return chunks_in_flight_; }; + + /// Return the number of chunks remaining. For testing only. + int64_t NumChunksRemaining() const { + int total = 0; + for (const auto &pair : push_info_) { + total += pair.second->chunks_remaining; + } + return total; + } + + /// Return the number of pushes currently in flight. For testing only. + int64_t NumPushesInFlight() const { return push_info_.size(); }; + + std::string DebugString() const { + std::stringstream result; + result << "PushManager:"; + result << "\n- num pushes in flight: " << NumPushesInFlight(); + result << "\n- num chunks in flight: " << NumChunksInFlight(); + result << "\n- num chunks remaining: " << NumChunksRemaining(); + result << "\n- max chunks allowed: " << max_chunks_in_flight_; + return result.str(); + } + + private: + /// Tracks the state of an active object push to another node. + struct PushState { + /// The number of chunks total to send. + const int64_t num_chunks; + /// The function to send chunks with. + const std::function chunk_send_fn; + /// The index of the next chunk to send. + int64_t next_chunk_id; + /// The number of chunks remaining to send. Once this number drops + /// to zero, the push is considered complete. + int64_t chunks_remaining; + + PushState(int64_t num_chunks, std::function chunk_send_fn) + : num_chunks(num_chunks), + chunk_send_fn(chunk_send_fn), + next_chunk_id(0), + chunks_remaining(num_chunks) {} + }; + + /// Called on completion events to trigger additional pushes. + void ScheduleRemainingPushes(); + + /// Pair of (destination, object_id). + typedef std::pair PushID; + + /// Max number of chunks in flight allowed. + const int64_t max_chunks_in_flight_; + + /// Running count of chunks in flight, used to limit progress of in_flight_pushes_. + int64_t chunks_in_flight_ = 0; + + /// Tracks all pushes with chunk transfers in flight. + absl::flat_hash_map> push_info_; +}; + +} // namespace ray diff --git a/src/ray/object_manager/test/push_manager_test.cc b/src/ray/object_manager/test/push_manager_test.cc index 82ae167ba..dd43d422e 100644 --- a/src/ray/object_manager/test/push_manager_test.cc +++ b/src/ray/object_manager/test/push_manager_test.cc @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -#include "ray/object_manager/object_manager.h" +#include "ray/object_manager/push_manager.h" #include "gtest/gtest.h" #include "ray/common/test_util.h" @@ -22,14 +22,15 @@ namespace ray { TEST(TestPushManager, TestSingleTransfer) { std::vector results; results.reserve(10); - UniqueID push_id = UniqueID::FromRandom(); + auto client_id = NodeID::FromRandom(); + auto obj_id = ObjectID::FromRandom(); PushManager pm(5); - pm.StartPush(push_id, 10, [&](int64_t chunk_id) { results[chunk_id] = 1; }); + pm.StartPush(client_id, obj_id, 10, [&](int64_t chunk_id) { results[chunk_id] = 1; }); ASSERT_EQ(pm.NumChunksInFlight(), 5); - ASSERT_EQ(pm.NumChunksRemaining(), 5); + ASSERT_EQ(pm.NumChunksRemaining(), 10); ASSERT_EQ(pm.NumPushesInFlight(), 1); for (int i = 0; i < 10; i++) { - pm.OnChunkComplete(); + pm.OnChunkComplete(client_id, obj_id); } ASSERT_EQ(pm.NumChunksInFlight(), 0); ASSERT_EQ(pm.NumChunksRemaining(), 0); @@ -39,21 +40,74 @@ TEST(TestPushManager, TestSingleTransfer) { } } +TEST(TestPushManager, TestSuppressDuplicates) { + std::vector results; + results.reserve(10); + auto client_id = NodeID::FromRandom(); + auto obj_id = ObjectID::FromRandom(); + PushManager pm(5); + + // First send. + pm.StartPush(client_id, obj_id, 10, [&](int64_t chunk_id) { results[chunk_id] = 1; }); + // Duplicates are all ignored. + pm.StartPush(client_id, obj_id, 10, [&](int64_t chunk_id) { results[chunk_id] = 2; }); + ASSERT_EQ(pm.NumChunksInFlight(), 5); + ASSERT_EQ(pm.NumChunksRemaining(), 10); + ASSERT_EQ(pm.NumPushesInFlight(), 1); + for (int i = 0; i < 10; i++) { + pm.StartPush(client_id, obj_id, 10, [&](int64_t chunk_id) { results[chunk_id] = 2; }); + pm.OnChunkComplete(client_id, obj_id); + } + ASSERT_EQ(pm.NumChunksInFlight(), 0); + ASSERT_EQ(pm.NumChunksRemaining(), 0); + ASSERT_EQ(pm.NumPushesInFlight(), 0); + for (int i = 0; i < 10; i++) { + ASSERT_EQ(results[i], 1); + } + + // Second allowed send. + pm.StartPush(client_id, obj_id, 10, [&](int64_t chunk_id) { results[chunk_id] = 3; }); + for (int i = 0; i < 10; i++) { + pm.OnChunkComplete(client_id, obj_id); + } + ASSERT_EQ(pm.NumChunksInFlight(), 0); + ASSERT_EQ(pm.NumChunksRemaining(), 0); + ASSERT_EQ(pm.NumPushesInFlight(), 0); + for (int i = 0; i < 10; i++) { + ASSERT_EQ(results[i], 3); + } +} + TEST(TestPushManager, TestMultipleTransfers) { std::vector results1; results1.reserve(10); std::vector results2; results2.reserve(10); - UniqueID push1 = UniqueID::FromRandom(); - UniqueID push2 = UniqueID::FromRandom(); + auto client1 = NodeID::FromRandom(); + auto client2 = NodeID::FromRandom(); + auto obj_id = ObjectID::FromRandom(); + int num_active1 = 0; + int num_active2 = 0; PushManager pm(5); - pm.StartPush(push1, 10, [&](int64_t chunk_id) { results1[chunk_id] = 1; }); - pm.StartPush(push2, 10, [&](int64_t chunk_id) { results2[chunk_id] = 2; }); + pm.StartPush(client1, obj_id, 10, [&](int64_t chunk_id) { + results1[chunk_id] = 1; + num_active1++; + }); + pm.StartPush(client2, obj_id, 10, [&](int64_t chunk_id) { + results2[chunk_id] = 2; + num_active2++; + }); ASSERT_EQ(pm.NumChunksInFlight(), 5); - ASSERT_EQ(pm.NumChunksRemaining(), 15); + ASSERT_EQ(pm.NumChunksRemaining(), 20); ASSERT_EQ(pm.NumPushesInFlight(), 2); for (int i = 0; i < 20; i++) { - pm.OnChunkComplete(); + if (num_active1 > 0) { + pm.OnChunkComplete(client1, obj_id); + num_active1--; + } else if (num_active2 > 0) { + pm.OnChunkComplete(client2, obj_id); + num_active2--; + } } ASSERT_EQ(pm.NumChunksInFlight(), 0); ASSERT_EQ(pm.NumChunksRemaining(), 0);