[object_buffer_pool] reduce mutex lock scope in WriteChunk

[sjoshi6]

Why are these changes needed?

Object store network transfer performance is slow, and we observe a periodic burst followed by a gap in the network usage.

A burst of inbound network traffic occurs at the beginning of each ray.get(obj_refs) call, then there is a wide-gap of unused network traffic and then a subsequent burst of network traffic in the next ray.get(obj_refs) call.

This looks like a processing bottleneck when payloads are received over the network on the Pull side.

We dug into the code in object_manager and found the following:

• Objects are transferred in chunks of size: 5 MiB
• When a PushRequest is received for a chunk, it is processed by ObjectManager::HandlePush
• Which internally calls the ObjectManager::ReceiveObjectChunk. This results in a call to the function ObjectBufferPool::WriteChunk.
• The WriteChunk function is mutex guarded throughout its execution.
• This includes the std::memcpy call for a 5MiB payload
• This pool_mutex_ lock is shared by all object_ids being received over the network

Which makes us believe that even if chunks for different ObjectId are received in parallel over the network they are written sequentially. Which would explain why we see a burst in the network usage followed by a hole in the network usage

Changes

Write Chunk

• Transition the chunk from REFERENCED to SEALED before releasing the lock
• Increment / Decrement num_inflight_copies before / after the copy
• Perform an unguarded memcpy of the chunk into the buffer
• Reacquire the mutex lock and perform object_id level Seal and Release decisions

AbortCreate

• Wait to ensure num_inflight_copies == 0 before allowing the Release & Abort calls for the object_id
• This check ensures that we do not release the underlying buffer while an unguarded copy is ongoing

Tests

Before

• Sampled network at 1 second frequency

speedometer.py -i 1 -m 64424509440 -n 1073741824 -rx eth0

• Sampled network at 100 millisecond frequency

speedometer.py -i 0.1 -m 64424509440 -n 1073741824 -rx eth0

After

• Sampled network at 1 second frequency
  speedometer.py -i 1 -m 64424509440 -n 1073741824 -rx eth0

• Sampled network at 100ms frequency
  speedometer.py -i 0.1 -m 64424509440 -n 1073741824 -rx eth0

Finished benchmark for total_size_MiB: 102400, block_size_MiB: 1024, parallel_block: None
	ray.wait(fetch_local=True) Gbps: 54.67178658153866
	ray.wait(fetch_local=True) total time s: 15.711823463439941
	ray.get() Gbps: 186301.24111362518
	ray.get() total time s: 0.004610776901245117

Related issue number

Checks

• I've signed off every commit(by using the -s flag, i.e., git commit -s) in this PR.
• I've run scripts/format.sh to lint the changes in this PR.
• I've included any doc changes needed for https://docs.ray.io/en/master/.
  • I've added any new APIs to the API Reference. For example, if I added a
    method in Tune, I've added it in doc/source/tune/api/ under the
    corresponding .rst file.
• I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
• Testing Strategy
  • Unit tests
  • Release tests
  • This PR is not tested :(

[sjoshi6]

@iycheng , @kevin85421 , @jjyao