[air] pyarrow.fs persistence (15/n): Trainable and Train worker artifact uploading

[justinvyu]

Why are these changes needed?

This PR enables artifact syncing on the Trainable (trial driver) and Train remote workers:

• Artifact uploading happens as a background task now on a result report, rather than a synchronous upload on a checkpoint report.
• sync_on_checkpoint is now used (previously it was only used in the head node syncing case). It forces a blocking artifact sync to happen on trial checkpoints.
• sync_artifacts=True by default still -- to be updated in a follow-up.

Related issue number

#38865

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

[justinvyu]

sync_on_checkpoint forces a blocking artifact sync to happen on trial checkpoints.

@ericl Should this instead force a WAIT on the previous task, then schedule a new one, and not necessarily block on the new one to finish?

-- wait on previous sync --> | launch new one | ---> let training continue

vs.

-- wait on previous sync --> | launch new one | --- wait on the new one --> | let training continue

[ericl]

-- wait on previous sync --> | launch new one | --- wait on the new one --> | let training continue

Unfortunately I think we have to do this one, otherwise you might miss some recently written files.

[justinvyu]

The code should be blocked at this point though, as we're waiting on the previous sync to finish up. What files are going to be missed?

[ericl]

The code should be blocked at this point though, as we're waiting on the previous sync to finish up. What files are going to be missed?

There's no guarantee the previous sync actually picked up a file that was written just before we called report. The sync could have started long before that point.

[justinvyu]

If we first wait for the old sync, then launch a new sync in report, I think we should be uploading the most up to date artifacts.

[zhe-thoughts]

Necessary change for Train