Discussion on batch Garbage Collection. #2242
Comments
|
We used the task_id(part of the objectID) to GC objects for a task in ant finance internal version, which gets very promising results for the plasma object management. I am thinking to migrate the feature to the open source version, so we raise the issue. @pcmoritz @robertnishihara , do you have any comments on it? |
|
Can you post a code example showing what this would look like from the application level? Or would this be completely under the hood? |
|
I will post a code review as an example shortly. Thank you Robert. |
|
Thanks! I have some reservations about this approach and how general it is. It clearly makes sense for at least one of your use cases, but we need to make sure we have a strategy that works in general. |
|
Here is the user code change that our scenario needs. alipay#29 |
|
In the user side, here is a sample of the user code. The user wants to do the GC to release the Object memory for the finished jobs. public RayJobScheduler {
public void init(JobSchedulerContext jobSchedulerContext) {
// Set the job id. Do some init work...
}
public void scheduleJob(IRayaJobDriver driver) {
// Set the job id. Do some init work.
Ray.startBatch(jobid,...);
}
public void finishJob(Long jobId, JobRes jobRes) {
// Do some cleaning work...
Ray.endBatch(...);
if (job.Res.doGc) {
Ray.gc(jobId);
}
}
}Here, we are referring JobId and BatchId as the same concept which means a whole job/procedure/experiment that generating many ray tasks and objects to achieve a goal. And the job finishes and we have the result, we know that the resources in the job are no longer needed, so we can release the resources for later and continuous job requirements. |
|
For the python code, user application will be the following code. The jobID is an option one, it will not affect the existing feature/code. Only user want to manage the task and object life cycle by himself, he can use the start_batch and end_batch to do GC. @ray.remote ray.init() jobID = ray.start_batch(); |
|
Hey @guoyuhong @eric-jj - I think we have been conservative in two things
So I think it'd be great to have a deeper discussion on this proposed feature, perhaps in the form of a design doc? @robertnishihara @pcmoritz what do you guys think? To clarify, we've been working on addressing GCS memory issue as well. The difference is our solution's more transparent to the user, while still being possible to call directly by the user should she choose so. Each GC batch is not tied to a job. When it's ready, I'd love to get you guys to play with this feature and see if it meets your needs! |
|
@concretevitamin I am trying to merge the internal version ray and github version. The first step to me is to deploy the github ray to our scenario. The feature is a blocking one for it, we can't apply the github version ray to our online system. @guoyuhong will write a design doc and we can follow up the detail design. To the design, it will introduce a top level API, it is expected. Some scenarios need the RAII ability, user can choose to use it or not. And I don't think it will exposing internal control to user, how to implement the GC logical is transparent to user, we can do nothing although user mentioned that. What's your plan to resolve the GCS memory issue, can you provide more details like design doc. We can evaluate it resolve our problem or not. |
BackgroundGraph ComputingWe have graph computing scenario based on Ray. Ray can be used as the backend for the continuous graph computing requests. The requests have the following characteristics:
Currently, the Plasma Store will evict the objects according to creation time when the memory is not enough. However, this mechanism is not enough. For one thing, our scenario needs large amount of memory, so it won't take long to use up the memory with incoming requests. For another, if an object is evicted according to creation time while it is still in use for later tasks, there will be a reconstruction of the object, which will affect the performance. Since we have the knowledge which requests are finished and won't be used in the future, we can release the resource safely. Streaming ComputingIn the further evolution, Ray will support stream computing. The life cycle for streaming objects are different. Even the best automatic garbage collecion may not be enough. We need to provide the elaborate memory management like RAII to enable users to manage object life cycle according to their needs. SolutionWe introduce a concept of Batch and BatchId ( Implementation DesignAPI Level# APIs:
# This function can be called after Ray.init().
# If this function is called, there is a batch concept in this Ray Driver.
# If not, Ray will work as normal.
ray.start_batch(batch_id)
# This function can be called by the users
# if they think all their resources are not needed in the future.
ray.end_batch(batch_id, need_gc = true)
###########################################################
# Scenario 1: Normal Mode
@ray.remote
def f():
time.sleep(1)
return 1
ray.init()
results = ray.get([f.remote() for i in range(4)])
# Scenario 2: Batch Mode
@ray.remote
def f():
time.sleep(1)
return 1
ray.init()
batchId = ray.start_batch()
results = ray.get([f.remote() for i in range(4)])
# Some other work...
ray.end_batch(batchId, need_gc = true)The Java API can be design in a similar way. ObjectId BitsCurrently, the lowest 32 bits are used for Object Index. The full higher 128 bits are used for TaskId. We can further split the 128 bits into two parts. The highest 64 bits can be used for BatchId. Backend SupportGCS SupportThere will be a GCS table to record the ObjectIds and TaskIds under one BatchId if the Batch mode is enable. The table name is "GC:${batchId}". Every objectId and taskId will be pushed into this table. After garbage collection is called with batchId, the GCS will release all the objectIds and taskIds under this table. Plasma SupportNode manager will subscribe the delete message of the "GC:${batchId}" from GCS. If GCS has deleted the garbage collection table, the node manager will trigger a garbage collection request to Plasma. Plasma will go through all the objectIds and delete the objects under the requested batchId. |
|
@robertnishihara @concretevitamin any comments? |
|
@guoyuhong thanks for the code example! @eric-jj, thanks, I am still thinking through the implications. Is there a clean way to implement this as a helper utility outside of the backend? E.g., in Python/Java you can track all of the object IDs that are generated between The advantage of doing it this way is that we can avoid adding functionality to the backend and also avoid adding extra information to the |
|
Is it accurate to say that the core issue is that the object store eviction policy is not good enough for some scenarios like you're streaming use case and that you want to provide application-level knowledge to improve the eviction strategy? |
|
@robertnishihara your solution can solve some problems, but it can't meet the requirments.
|
@robertnishihara extactly |
|
I think currently Ray is still lack of a good resource management ability. Thus, I think we can consider introducing a Resource Manager component into the core. It takes responsibility for managing the life cycle of all resources in the system, including GCS, object store, etc. I haven't thought about the detailed design of the Resource Manager. But I think having a component that unifies resource management will be great. Also, I'd like to hear @concretevitamin's plan to address GCS memory issue. |
|
@robertnishihara For the proposal, it is not only for the GC management also. It is a starting point for the job level management. When I talked with ant finance's AI team, they have strong feature requirement for the job level management, such as observe the job status, cancel or re-submit a job. |
|
@guoyuhong thanks for raising these issues. These are great points to discuss. First, regarding a job API. What are the requirements? Something like Spark submit (https://jaceklaskowski.gitbooks.io/mastering-apache-spark/spark-submit.html)? What else? Second, regarding object eviction from the object store, it would be great to understand when LRU is not good enough. What workload (it looks to me that for the graph workload LRU should be fine, but maybe I'm missing something) ? Of course, by LRU I mean that the objects will be evicted according with the times since they've been last accessed (red/write). And if we have workloads for which LRU is bad, one question here is whether configuring the eviction policy per driver/job would be good enough. Also, related to the above we need to add the ability to spill over the objects to a persistent store. If you guys have some thoughts here please let us know. Also, this would be a great task to collaborate or maybe for you to lead. |
|
@istoica For the job API, the Spark submit is a similar one. And Ray should provide a http server to user can submit, view, cancel the jobs. Another required important feature about job is separate resource manager for different user to avoid one user occupy all the resources in a cluster. If the feature(resource isolation) is not a general one, we need the interface in the job management module to let us apply our internal policy. For the LRU, different job's life cycle is different, many small jobs create many temp variables and exit, and few large job run a long time. So the objects created by the long run job will be created will be deleted because its' life time is longer than the objects' created by the small jobs. We did consider to create a new GC algorithm to resolve the issue. Ray has provide the interface, it will modify the code less. But we can't find a good solution to our scenarios. LRU is a GC algorithm, the GC algorithm is not better than java GC algorithm. It can't provide the scope resource management just like RAII in c++ and using in C#. If you have concern on add top level API for RAY, with the job level support, we can add a flag to submit to let user decide recycle the objects created in the job or not. It is more nature to user, and will not expose the details in the job. Spill over the objects to a persistent store is an option, but it can't resolve the issue because many temp variables are useless when a job exit in graph computation. The objects inflation can't be avoided in graph computation. In our case, 1 node will access the 300 neighbor nodes, it will be done for 3 times, which means 300^3 temp variables will be created. After a job exit, we don't want to maintain these objects. And in our current thinking, spill over the objects to persistent store need job level support. Retired objects in a job will be put to the persistent store as a group, and we will not keep entries for these retired objects in the plasma store. We also are trying to apply the ray graph to online calculation scenarios now, the perf of it is very sensitive, we only have less than 300ms budget, large page out and page in to persistent store is not acceptable. |
|
@istoica LRU is not enough because the storage unnecessarily caches a lot of objects which brings in significant memory/storage/computation overhead in our long running service case with small jobs. We therefore introduce the above batch based approach in our internal version, and it is now running in production. However, we do find limitations though for this approach as it does not allow interactions among different jobs (e.g., requests), and it does not support big job nicely (e.g., when the workload does not fit into total memory capacity). We therefore consider new distributed garbage collection solutions as mentioned in Tahoe to @robertnishihara etc. We will come back to you later when it is validated in production. Said so, we still think a batch based approach valuable and it is useful for ray to come up the new capability for being a long running job service quickly. |
|
@eric-jj and @imzhenyu thanks for the clarifications. A few things:
Also, one question here. How useful would be an API like (I know that the above API might not be as easy to use, but certainly is very flexible.) An alternative would be to provide an API like |
|
@istoica I will follow up the Ray Submit issue next week. |
|
Thanks, @eric-jj. BTW, it would still be great to understand the workloads for which doesn't LRU work well. Maybe there is an issue with the current implementation, rather than a LRU-policy problem? Also, in addition to evicting all objects of a job when finishes execution, would pinning an object in the object store be good enough? Arguably pinning would be easier to implement than ray.unsafe_evict(), and also more consistent with memory management patterns in OSes: https://lwn.net/Articles/600502/ |
|
@istoica |
|
@eric-jj, thanks for clarification. So I assume that in this case, during B's executions, A's won't access its data too much. Let's then design a job API, and free resources under the hood when the job ends. This solution would require no Here is a possible API for Python: Alternatively, we can define a class/actor that also has methods to get the job progress. Maybe this is the way to go. Note that under the hood, we could use the So it seems to me that the most important next steps are to define the job concept (e.g. what are its properties, e.g., name, unique id, scheduling parameters, ability to get a progress report, ...), and the API, including a submit API. Anyone would like to start a PRD on this? |
|
@istoica Yes. Job level management is more make sense. We introduce the batch because we it will change less code. As I mentioned before, I planned to do it step by step. Implement the batch first, then build job on it. |
|
Sounds good @eric-jj. I'help with the PRD document. It would be great to do it over the next couple of days. Thanks. |
|
A different API I'd propose here is to use context managers for this: Usage with the @ray.remote
def f(something):
# Do something
with ray.job(name="jobname"):
x = f.remote(1)
# Do other things
# The object "x" can now be cleaned up
with ray.job(name="jobname2"):
y = f.remote(1)
# Do moreContext managers can also be used without the The semantics here would be that there is one default job per driver and you can create custom ones that inherit all objects and function definitions from that default job. It is not allowed to pass objects between these inner jobs, so after the context manager is closed on the driver and all tasks from that job have finished, things can be cleaned up. |
|
@pcmoritz yes, with operation seems more reasonable for the scenario. @guoyuhong please add it to the design doc. |
|
@istoica For this pattern: # Actor 1 -----S1.1----------------S1.2---------|--- -> timeline
# ^ ^ |
# | T1.1 | T1.3 (finish) |
# Batch 1 --------------------------------------|--- -> timeline
# | T1.2 |
# v |
# Actor 2 ---------S2.1----S2.2-----------------|--- -> timeline
# ^ |
# | T2.1 |
# Batch 2 --------------------------------------|--- -> timeline
# |
# (cleanup moment of batch 1)
On the other words, when If I misunderstand the scenario, please point it out. Thanks a lot. |
|
@istoica regarding to the previous 2 questions about sync/async. (1). "other tasks may further emit more actor tasks" Yes, it is correct. (2). Yes, async API is better for streaming, IMHO. Implementing that with sync API is doable I guess, but it will introduce unnecessary delays and complications, because streaming is more similar to one way RPC, where the upstream nodes do not care much about what the downstream nodes will do. With a |
|
@istoica My previous statement about |
|
@imzhenyu can you provide an example of why we need multiple |
|
Basically, I'm thinking of something like this where Upon finishing the batch, the above code will clean up And, here is the |
|
@imzhenyu BTW, in the above example garbage collection happens only after the batch ends. Nothing will be garbage collected before then. So as long as you add an object to the return list, you can safely use that object elsewhere as soon as it is created as you know it won't be garbage collected during or after the batch ends (no need to wait for In the above example, say you do a lot of stuff after creating Wouldn't this be enough for your cases? |
|
@imzhenyu Maybe rename |
|
@yuyiming and @zhijunfu Thanks for your example illustrating that it is hard to checkpoint actors on the batch frontier. However, the proposal I added to the document should still work correctly. Is the a problem you have in mind an efficiency or correctness problem? Anyway, I was thinking about the following alternative that will make the above discussion moot: how about checkpointing the actors at the beginning of the batch instead of at the end? As long as the actors are checkpointed frequently enough it shouldn't matter whether we do it at the beginning or the end. In a streaming example, just think of the next batch checkpointing the state of the current batch. This would be much easier to implement and it will have the added benefit to enable rollback, e.g., in an SGD iteration you can rollback to the initial state of an iteration if an actor fails, which is what you want. |
|
@istoica Agree that each batch should only have one return statement (when I say As for the |
|
@imzhenyu can you please give an example showing this pattern, i.e., (1) needing to have a |
|
@istoica I don't have an immediate example for both. But for (1), considering a streaming example where the initial points only calls a function without a return value (as it is the ingest nodes), and the batch ends in a later stage deployed in another actor. For (2), considering caching the results for the computation requests, then you need to collect z a later point. |
|
@imzhenyu Thanks. Correct me if I'm wrong but for (1) I think that you are saying there is no Also, for (2) you are basically saying that you want to garbage collect an object outside the initial |
|
@istoica I think one solution is to checkpoint Actor states on all batch transition points (batch boundaries). For instance, When |
|
@istoica yes, there is no returned value or the batch returns empty results. (2) is correct. |
|
@yuyiming and @imzhenyu, given that there are still questions about the use cases and hence about the high level API, how about starting with a low-level API, use it, and build a higher level API as we get more experience with the use cases. So how about the following low-level API:
Please let me know if the above API could cover all your use cases. Again, the key question here is feasibility not convenience. Thanks. |
|
@istoica Agree, we can definitely go further first. I think @guoyuhong is already working on it :-) |
|
@imzhenyu sounds good. So let's then go with the above plan then. According to @stephanie-wang we already have functionality for (2) but we need to expose it. @robertnishihara is looking at (3). Regarding (4) we need to think a bit of semantics, e.g.,
Also, a better name for this API is |
|
@imzhenyu, @eric-jj, @yuyiming, @robertnishihara, @pcmoritz I added the above proposal to the end of our design document: https://docs.google.com/document/d/1sMgVMajomF5j-GDkzhuejrAwLcxXNKY9lTmlDo7GkqM/edit?ts=5b360317# Also wrote another small document about (4) here: https://docs.google.com/document/d/1kcl5SHXtQJQ5_F69rSRgbLrJ9LYDPEmY-totALi7BTc/edit# Please take a look and add any comments you might have. I really think we should start with implementing this proposal and use if for various example apps to get experience with it. Since (4) is the most complex, in the first phase I propose we do the first tree. |
|
@imzhenyu, @eric-jj, @yuyiming, @robertnishihara, @pcmoritz Just added some examples and updated the two documents: https://docs.google.com/document/d/1sMgVMajomF5j-GDkzhuejrAwLcxXNKY9lTmlDo7GkqM/edit?ts=5b360317# and https://docs.google.com/document/d/1kcl5SHXtQJQ5_F69rSRgbLrJ9LYDPEmY-totALi7BTc/edit# |
|
@istoica Sorry for the late response. Regarding to the semantic of kill & delete, I'm wondering it is better that a FO system does not have user-facing |
|
@imzhenyu Thanks for the note. Can you please be more precise about GC? When would you GC? Also, what about cancelling tasks/methods on timeouts? A few notes:
|
|
@istoica Hi, Ion, I'm implementing Moreover, since these functions are low-level functions, can we put them into another namespace to mark that these functions should be used carefully? For example, we can use |
|
@guoyuhong Can you please write down the exact algorithm you are planning to implement? When is the reference counter increased, and when it is decreased? What happen in case of failure? (As a disclaimer, we did consider reference counting early on, but we gave up since maintaining the counter in a distributed system in the presence of failures and message losses is complex; it's equivalent with implementing transactions in a distributed system, as you need to provide exactly once semantics.) Again, I think that just deleting the objects or marking them for deletion so you evict them first when the object store is full, would be a fine solution--simple and effective. This is because we do not delete the lineage from GCS (we only flush it) so you can still reconstruct the objects after removing them from the object store, if needed. Yes, using |
|
Current object store has implemented the ref count. Here is the code. The ref count is used to count the usage of an object by all the Plasma Clients. I have send a PR to cache the delete object ids, and these objects will be deleted once its ref count is 0. However, the L3 cache in PlasmaClient will held the Release request until the cache condition satisfies. Current condition is that the release cache has more than 64 items or cache memory reaches 100MB. Otherwise, this client will hold the reference of the target objects even it is released. My current object store free algorithm:
From my current experiments, this algorithm works. But we need to wait until there are more than 64 items pushed to the L3 cache of Plasma Clients. |
|
@guoyuhong Thanks for clarifications. I think there was a confusion. In my original post, I was referring to having global reference counters. The ones in Ray are per node. At the first sight, your solution makes sense. Also, I think that waiting for cache condition is not a big deal as 100MB is relatively tiny for today's machines. But we might want to change this, especially if we want to run Ray on edge devices as some users do. |
|
I don't think this is relevant anymore now that we're working on doing exact GC for |
Hi @robertnishihara @pcmoritz , we are planning to add a batch Garbage Collection to Ray.
We have a concept called batchId (int64_t) used to do the Garbage Collection. For example, one job will use this batchId to generate all the objectIds and taskIds, and all these objectIds and taskIds will be stored under the Garbage Collection Table under the batchId in GCS. When the job is finished, we can simply pass a batchId to the garbage collector and the garbage collector will look up the Garbage Collection table in GCS and do the garbage collection to all the related tasks and objects.
In current id.h implementation, the lowest 32 bits in ObjectId is used for Object Index. We can use the higher 64 bits next to the Object Index as the batchId and add a new GC Table in GCS.
This GC mechanism will help release the memory resources in GCS and plasma.
How do you think of this code change?
The text was updated successfully, but these errors were encountered: