Confusing behavior of `Context.from_options` and statistics

[TomAugspurger]

After #1009, calling Context.from_options with an Options enabling statistics doesn't result in a Context with enabled statistics.

import rapidsmpf.config
import rapidsmpf.statistics
import rapidsmpf.streaming.core.context
import rapidsmpf.communicator.single
import rapidsmpf.progress_thread
import rapidsmpf.rmm_resource_adaptor
import rmm


# confusing behavior: Context.from_options() doesn't honor statistics=True in Options;

opts = rapidsmpf.config.Options({"statistics": "True"})

comm = rapidsmpf.communicator.single.new_communicator(
    progress_thread=rapidsmpf.progress_thread.ProgressThread(),
    options=opts,
)
mr = rapidsmpf.rmm_resource_adaptor.RmmResourceAdaptor(rmm.mr.get_current_device_resource())

context = rapidsmpf.streaming.core.context.Context.from_options(
    comm.logger,
    mr,
    opts
)
assert context.statistics().enabled  # AssertionError

you need to instead pass in a Statistics, likely built from Statistics.from_options(opts):

context = rapidsmpf.streaming.core.context.Context.from_options(
    comm.logger,
    mr,
    opts,
    rapidsmpf.statistics.Statistics.from_options(opts)

)
assert context.statistics().enabled  # True

Having two ways to specify the statistics is confusing. I'd recommend one or the other: rapidsmpf should consistently use the options anywhere a statistics might be created / used, or it should not have a statistics option.

[nirandaperera]

The reason I had to do this was, by the time we create a Context, it's likely that we've created a communicator/ progress theeat which needs a stats object. We need to use this everywhere. Since buffer resource and context dont require a communicator or progress thread, we need to pass it externally.

[nirandaperera]

Maybe we need a Rapidsmpf runtime environment which has a from_options method that handles all of this transparently and use it everywhere.

[nirandaperera]

Related to this #1017
This introduces a Statistics::reset_from_options method, that resets stats inplace based on new options map.
rapidsai/cudf#22690 uses it.

[madsbk]

The statistics argument will be mandatory everywhere, so you will have to do something like this to create the ProgressThread:

stats = rapidsmpf.statistics.Statistics.from_options(opts)
rapidsmpf.progress_thread.ProgressThread(statistics=stats)