# DAGDriver memory increasing for basic DAGs

**URL:** <https://discuss.ray.io/t/dagdriver-memory-increasing-for-basic-dags/8758>\
**Category:** Ray Serve\
**Created:** [December 22, 2022, 3:35pm UTC](https://discuss.ray.io/t/dagdriver-memory-increasing-for-basic-dags/8758 "2022-12-22T15:35:49Z")\
**Posts on this page:** 6\
**Page:** 1

<div class="post-metadata">

**Author:** ![AbhishekBose](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.ray.io/abhishekbose/32/3292_2.png) [@AbhishekBose](https://discuss.ray.io/u/AbhishekBose)\
**Post date:** [December 22, 2022, 3:35pm UTC](https://discuss.ray.io/t/dagdriver-memory-increasing-for-basic-dags/8758/1 "2022-12-22T15:35:49Z")

</div>

Hello Team,

I am testing out DAG driver for creating parallel processing branches for one of our use-case.  
The use-case is as follows:

1. Base64 image comes as request
2. Image is converted to proper input for model
3. External api call (Azure ocr api) is performed in one deployment
4. Object detector model (Detectron) is called in another deployment.

I have implemented the functionality using deployments and DAGdriver. Although functionally it has been working as expected. I was seeing a constant memory increase for continuous requests.

In order to investigate properly, I converted my processing deployments (AzureAPI call and Detectron model inference) into simple functions which just return a dictionary or arrays.

## DetectronDeployment

```auto
@serve.deployment()
class DetectronDeployment:
    def __init__ (self):
        self.doc_detector = ObjDetector(MODEL_CFG)
        self.doc_processor = DocDetector(640, 640)

    async def detect_doc(self, file_bytes):
        start_time = time.time()
        image_np = np.frombuffer(file_bytes, np.uint8)
        img = cv2.imdecode(image_np, flags=1)

        try:
            img_scaler = self.doc_processor.standardise_img(img)
            detections = self.doc_detector.get_pred(self.doc_processor.final_img)

        except Exception as ex:
            log.error(ex)
            log.error(traceback.print_exc())

        filter_detections = []
        for i in range(len(detections)):
            filter_detections.append(
                {
                    "doc_type": self.doc_detector.classes[
                        detections.pred_classes[i].item()
                    ],
                    "confidence": detections.scores[i].item(),
                    "bbox": self.doc_processor.process_bbox(detections.pred_boxes[i]),
                }
            )
        log.info("Time taken for detections :: {}".format(time.time() - start_time))

        scale_params = self.doc_processor.export_params()

        return json.dumps([scale_params, filter_detections])

    async def forward(self,uploaded_file):
        start_time = time.time()
        resp = {
            "detection values":[1,2,3,4,5]
        }
        # resp = await self.detect_doc(uploaded_file)
        log.info("Time taken to analyze image :: {}".format(time.time() - start_time))
        return resp

detectron_deployment = DetectronDeployment.bind()

```

## AzureDeployment

```auto
@serve.deployment()
class AzureDeployment:
    def __init__ (self):
        self.ocr_api = AzureAPI(os.environ.get("SUBKEY"), os.environ.get("AZURE_ENDPOINT"))

    def forward(self,uploaded_file):
        start_time = time.time()
        time.sleep(0.2)
        # resp = self.ocr_api.analyze_image(uploaded_file)
        resp = {
                "status": "success",
                "status_code": 200,
                "result": {
                    "extracted_data": {
                        "aadhaarNumber": "567420078076",
                        "dob": " 08/01/1991",
                        "gender": "Male",
                        "name": "Purushothama",
                        "fatherName": "NA",
                        "address": "NA",
                        "docType": "aadhar front half"
                    },
                    "image_quality": {
                        "blur": 0,
                        "res": 0.05
                    }
                },
                "version": "1.2"
            }
        log.info("Time taken to analyze image :: {}".format(time.time() - start_time))
        return resp

azure_deployment = AzureDeployment.bind()

```

As you can see both the deployments contain a forward function which basically does nothing more than returning an array and a json respectively

My DAGDriver graph is defined in this way

```auto
with InputNode() as input_file:
    print(f"input file is :: {input_file}")
    health_response = health_deployment.forward.bind()
    dag = aggregate.bind(azure_deployment.forward.bind(input_file),detectron_deployment.forward.bind(input_file))

serve_dag = DAGDriver.options(num_replicas=1).bind({
    "/process":dag,
    "/health": health_response
})

```

Here is health\_deployment is →

```auto
@serve.deployment()
class HealthDeployment():
    async def forward(self) -> str:
        return "Server is healthy"

```

If you look at the screenshots below, you will find that that though the object store memory and the memory utilised per Deployment is constant, but still the DAGDriver replica linearly increases with the requests coming in

### Timestamp 1

 ![Screenshot 2022-12-22 at 8.51.14 PM](https://us1.discourse-cdn.com/flex020/uploads/ray/original/2X/0/0c3cf6b27ca6177406f368bfe094af11d550a2ff.png)

### Timestamp 2

 ![Screenshot 2022-12-22 at 8.51.57 PM](https://us1.discourse-cdn.com/flex020/uploads/ray/original/2X/2/223e73a5e327a243d8e460da14304c21dfd87a50.png)

### Timestamp 3

 ![Screenshot 2022-12-22 at 8.53.15 PM](https://us1.discourse-cdn.com/flex020/uploads/ray/original/2X/b/bb76d9a601cd0202a55eab08e974ce895936f565.png)

### Timestamp 4

 ![Screenshot 2022-12-22 at 9.03.38 PM](https://us1.discourse-cdn.com/flex020/uploads/ray/original/2X/4/41fb54f0042154b1d58976b31ae7fd88eb885213.png)

Kindly have a look at this and let me know if I am doing something wrong here with my deployments and request flow

---

<div class="post-metadata">

**Author:** ![simon-mo](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.ray.io/simon-mo/32/9_2.png) [@simon-mo](https://discuss.ray.io/u/simon-mo)\
**Post date:** [December 22, 2022, 5:25pm UTC](https://discuss.ray.io/t/dagdriver-memory-increasing-for-basic-dags/8758/2 "2022-12-22T17:25:15Z")

</div>

Hi @AbhishekBose, would you mind share which Ray version are you using? And additionally, what does the request traffic look like? Do you have a continuous stream of queries keep piling up to be processed? One potential cause of memory increase is the requests being queued up in memory in the DAGDriver process. We would also appreciate a reproduction script that is minimal so we can help diagnose and debug as well!

---

<div class="post-metadata">

**Author:** ![AbhishekBose](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.ray.io/abhishekbose/32/3292_2.png) [@AbhishekBose](https://discuss.ray.io/u/AbhishekBose)\
**Post date:** [December 23, 2022, 9:43am UTC](https://discuss.ray.io/t/dagdriver-memory-increasing-for-basic-dags/8758/3 "2022-12-23T09:43:37Z")

</div>

@simon-mo  
Ray-version → 2.2.0  
python version → 3.8.16

The test was done using locust. So the program kept sending http requests consisting of base64 image string to the ray sever.

Here is the testing code as well:

```auto
from locust import HttpUser, task, constant
import json
import random
import os

URL = "http://0.0.0.0:8000/process"

class RayServeTest(HttpUser):
    # wait_time is compulsory, even if its 0,
    wait_time = constant(0)

    @task
    def hit1(self):
        self.client.post(url=URL,json={
            "image":"base_64_string"
            }

```

> **[GitHub - AbhishekBose/ray-serve-dag-experiment: Contains experiments with ray...](https://github.com/AbhishekBose/ray-serve-dag-experiment)**
>
> Contains experiments with ray serves new Dag module - GitHub - AbhishekBose/ray-serve-dag-experiment: Contains experiments with ray serves new Dag module

Here is a public git repo which you guys try out.

```auto
docker compose up

```

should start the server.

---

<div class="post-metadata">

**Author:** ![AbhishekBose](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.ray.io/abhishekbose/32/3292_2.png) [@AbhishekBose](https://discuss.ray.io/u/AbhishekBose)\
**Post date:** [December 23, 2022, 5:53pm UTC](https://discuss.ray.io/t/dagdriver-memory-increasing-for-basic-dags/8758/4 "2022-12-23T17:53:32Z")

</div>

@simon-mo Kindly have a look at the repo and suggest what could be done about this. Are we doing something wrong on our end or is the issue related to the package itself?

---

<div class="post-metadata">

**Author:** ![simon-mo](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.ray.io/simon-mo/32/9_2.png) [@simon-mo](https://discuss.ray.io/u/simon-mo)\
**Post date:** [December 23, 2022, 8:17pm UTC](https://discuss.ray.io/t/dagdriver-memory-increasing-for-basic-dags/8758/5 "2022-12-23T20:17:06Z")

</div>

I can reproduce this in Ray 2.2.0 (as well as Ray 2.0.1, 2.1.0)

---

<div class="post-metadata">

**Author:** ![simon-mo](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.ray.io/simon-mo/32/9_2.png) [@simon-mo](https://discuss.ray.io/u/simon-mo)\
**Post date:** [December 23, 2022, 8:39pm UTC](https://discuss.ray.io/t/dagdriver-memory-increasing-for-basic-dags/8758/6 "2022-12-23T20:39:15Z")

</div>

Fix PR is here: [[Serve] Address incremental memory leak due to \_PyObjScanner by simon-mo · Pull Request #31317 · ray-project/ray · GitHub](https://github.com/ray-project/ray/pull/31317)

Once the PR is merged, you can install the Ray nightly wheel to confirm. [Installing Ray — Ray 3.0.0.dev0](https://docs.ray.io/en/master/ray-overview/installation.html#daily-releases-nightlies)

Edit: PR merged!