[core] Eagerly evict objects that are no longer in scope

[stephanie-wang]

Why are these changes needed?

Add an option to eagerly evict copies of objects that are no longer in scope, according to the owner of the object, to reduce plasma's memory footprint. The eviction is done by the raylets. When the raylet that is pinning the object ID hears from the owner that it is OK to unpin, the raylet adds the object ID to a list of objects to free. Once the list reaches a configured size or has not been flushed after a configured time, the list will be flushed by sending a FreeObjects request to all other object managers. This in turn triggers a Delete of the object in plasma.

This PR leaves the feature off by default. To enable it, make sure object_pinning_enabled is on, then set free_objects_period_milliseconds to a non-negative value in the backend config. This will set the time period between attempts to free objects. free_objects_batch_size sets the maximum size before the list is flushed. Note that if the application uses serialized ObjectIDs, i.e. an object ID that is created in one process and another process is given a reference to it, then it is recommended that distributed_ref_counting_enabled is also turned on, or else the application may receive spurious "object lost" errors.

Example to enable:

ray.init(_internal_config=json.dumps({
    "free_objects_period_milliseconds": 1000,  # Flush objects every second.
    "free_objects_batch_size": 100,  # Flush if >=100 objects were unpinned locally since the last flush.
    "distributed_ref_counting_enabled": 1,  # Recommended for serialized IDs.
  }))

Checks

• I've run scripts/format.sh to lint the changes in this PR.
• I've included any doc changes needed for https://ray.readthedocs.io/en/latest/.
• I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failure rates at https://ray-travis-tracker.herokuapp.com/.

[AmplabJenkins]

Can one of the admins verify this patch?

[AmplabJenkins]

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/22095/
Test FAILed.

[edoakes]

Looks great!

Might've preferred first having a version that doesn't batch at all but given that it's implemented this seems good.

[edoakes]

Doing this in the heartbeat means the flush duration needs to be much larger than or a multiple of the heartbeat duration - might be better to just have a separate timer on the event loop. Otherwise should indicate this in the comments in the config.

[virtualluke]

object_pinning_enabled is on by default now?

Is there an api to add an object to the "list of objects to free" that will get evicted on the time or batch size threshold mentioned in this PR?

thanks for the work on this.

[stephanie-wang]

object_pinning_enabled is on by default now?

Yes, object_pinning_enabled is on by default, I believe as of the last release.

Is there an api to add an object to the "list of objects to free" that will get evicted on the time or batch size threshold mentioned in this PR?

You could call del object in Python and assuming that's the only reference left, that should automatically add the object to the list. If that doesn't work for you, you can also try the internal API ray.internal.free, which immediately attempts to free whichever objects you passed in. Since this API is internal, though, there may not be support for it in the future.

[AmplabJenkins]

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/22127/
Test FAILed.

[AmplabJenkins]

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/22135/
Test FAILed.

[stephanie-wang]

TestMemoryScheduling::testTuneWorkerHeapLimit failure looks like it's from master.