[RLlib] Make learner group with non blocking update return multiple results

[avnishn]

Signed-off-by: Avnish [email protected]

The results aggregation of the learner was broken becasue we upped the amount of async requests for the learner. This change should fix that problem.

Why are these changes needed?

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 :(

[kouroshHakha]

I think new API modifications are good, but can we clean up the learner_group a little bit? The names have become convoluted over time. I suggested some better names. Another suggestion:

update(block=True/False) --> update(async=True/False)

[kouroshHakha]

Why not store the tagged results in a dict mapping from tag to list of results? This gives you O(1) on the read.

processed_results = defaultdict(list)
for result:
     if ok:
           processed_results[result.tag].append(result_or_error)

Then

for tag in processed_results:
      self._inflight_request_tags.remove(tag)

[avnishn]

I can make this change but for the given amount of tags << 10000, this isn't going to make any significant difference to actual runtime. Since we aren't operating on a large amount of data here, I'll rewrite this in a way where its as easy to modify/maintain as possible.

[avnishn]

all done!

[kouroshHakha]

I think new API modifications are good, but can we clean up the learner_group a little bit? The names have become convoluted over time. I suggested some better names. Another suggestion:

update(block=True/False) --> update(async=True/False)

[kouroshHakha]

We need the API description to be higher-level and simpler to understand. It's explaining the code right now.

[avnishn]

we should just break this into 2 functions instead:

update, and async_update. Multiple return types is weird.

[kouroshHakha]

Ok let’s do that instead.

[sven1977]

Let's add a TODO then also for the reduce_fn. We have different return types here, too, because of this.
Result reduction should happen on the training_step side.

[avnishn]

@sven1977 done

[kouroshHakha]

The goal of this PR is merely fixing the test of learner_group, but learner_group's API surface is still not clean and can be simplified in another iteration, similar to what we did for Learner. Approving the merge for now as it's blocking CI's green-ness.

[ollie-iterators]

"If the queue is full itwill evict the oldest batch first." should be "If the queue is full it will evict the oldest batch first."