Use `arrow.array` instead of `PyBytes` to enables zero copy

[haixuanTao]

This implementation makes use of the newly introduce arrow.array to share messages between nodes and operators with zero-copy.

The effective changes are:

• replace sending output from using Pybytes to arrow array:

  # From
  outputs.tobytes(),
  # To
  pa.array(outputs.ravel().view(np.uint8))

  ravel() makes the array 1-Dimensional without copy. This also makes sure that the data is contiguous. .view(np.uint8) makes the array viewed as bytes.

• replace receiving input from Pybytes to arrow.array:

  # From
          self.position = np.frombuffer(dora_input["data"], np.float32)
  # To
          self.position = np.array(dora_input["value"]).view(np.float32)

  view() makes changing the dtype without copy

There is few

[meua]

This feature is really cool, I'm going to use it.

[meua]

This implementation makes use of the newly introduce arrow.array to share messages between nodes and operators with zero-copy.

The effective changes are:

• replace sending output from using Pybytes to arrow array:

# From
outputs.tobytes(),
# To
pa.array(outputs.ravel().view(np.uint8))

ravel() makes the array 1-Dimensional without copy. This also makes sure that the data is contiguous. .view(np.uint8) makes the array viewed as bytes.

• replace receiving input from Pybytes to arrow.array:

# From
            self.position = np.frombuffer(dora_input["data"], np.float32)
# To
            self.position = np.array(dora_input["value"]).view(np.float32)

view() makes changing the dtype without copy

There is few

pa.array(outputs.ravel().view(np.uint8))

Is the variable pa a typo? Are you using the abbreviation np of numpy?

[haixuanTao]

In the beginning of the file, you can see that import pyarrow as pa meaning that the import of pyarrow has been aliased to pa

[meua]

In the beginning of the file, you can see that import pyarrow as pa meaning that the import of pyarrow has been aliased to pa in webcam_op.py and dora version is v0.2.2

        ret, frame = self.video_capture.read()
        if ret:
            frame = cv2.resize(frame, (OUTPUT_WIDTH, OUTPUT_HEIGHT))
            frame = cv2.cvtColor(frame, cv2.COLOR_BGR2BGRA)

            send_output("image", pa.array(frame.ravel().view(np.uint8)), dora_input["metadata"])

Thank you, I did this, but the following data type conversion error is still reported.

(dora3.7) jarvis@jia:~/coding/rust_home/github.com/meua/dora-drives$ dora start graphs/tutorials/webcam.yaml --attach --hot-reload --name webcam
a5d6e10e-67b7-4e8a-b8d2-0e42bf976bb6
Traceback (most recent call last):
  File "<string>", line 1, in <module>
RuntimeError: Dora Runtime raised an error.

Caused by:
   0: main task failed
   1: operator webcam/op raised an error
   2: error in Python module at /home/jarvis/coding/rust_home/github.com/meua/dora-drives/operators/webcam_op.py
   3: TypeError: argument 'data': 'UInt8Array' object cannot be converted to 'PyBytes'Traceback (most recent call last):
        File "/home/jarvis/coding/rust_home/github.com/meua/dora-drives/operators/webcam_op.py", line 28, in on_event
          return self.on_input(dora_event, send_output)
        File "/home/jarvis/coding/rust_home/github.com/meua/dora-drives/operators/webcam_op.py", line 41, in on_input
          send_output("image", pa.array(frame.ravel().view(np.uint8)), dora_input["metadata"])

Location:
    binaries/runtime/src/operator/python.rs:26:9

[meua]

In the beginning of the file, you can see that import pyarrow as pa meaning that the import of pyarrow has been aliased to pa in webcam_op.py and dora version is v0.2.2

        ret, frame = self.video_capture.read()
        if ret:
            frame = cv2.resize(frame, (OUTPUT_WIDTH, OUTPUT_HEIGHT))
            frame = cv2.cvtColor(frame, cv2.COLOR_BGR2BGRA)

            send_output("image", pa.array(frame.ravel().view(np.uint8)), dora_input["metadata"])

Thank you, I did this, but the following data type conversion error is still reported.

(dora3.7) jarvis@jia:~/coding/rust_home/github.com/meua/dora-drives$ dora start graphs/tutorials/webcam.yaml --attach --hot-reload --name webcam
a5d6e10e-67b7-4e8a-b8d2-0e42bf976bb6
Traceback (most recent call last):
  File "<string>", line 1, in <module>
RuntimeError: Dora Runtime raised an error.

Caused by:
   0: main task failed
   1: operator webcam/op raised an error
   2: error in Python module at /home/jarvis/coding/rust_home/github.com/meua/dora-drives/operators/webcam_op.py
   3: TypeError: argument 'data': 'UInt8Array' object cannot be converted to 'PyBytes'Traceback (most recent call last):
        File "/home/jarvis/coding/rust_home/github.com/meua/dora-drives/operators/webcam_op.py", line 28, in on_event
          return self.on_input(dora_event, send_output)
        File "/home/jarvis/coding/rust_home/github.com/meua/dora-drives/operators/webcam_op.py", line 41, in on_input
          send_output("image", pa.array(frame.ravel().view(np.uint8)), dora_input["metadata"])

Location:
    binaries/runtime/src/operator/python.rs:26:9

Use the latest dora-rs/dora main branch, launch dora-deamon, dora-coordinator, and report the following error:

(dora3.7) jarvis@jia:~/coding/rust_home/github.com/meua/dora-drives$ dora start graphs/tutorials/webcam.yaml --attach --hot-reload --name webcam
Error: failed to deserialize incoming message

Location:
    binaries/cli/src/main.rs:223:44

[haixuanTao]

You can now try this version using the release candidate: https://github.com/dora-rs/dora/releases/tag/v0.2.3-rc6 :tada:

[meua]

You can now try this version using the release candidate: https://github.com/dora-rs/dora/releases/tag/v0.2.3-rc6 tada

(dora3.7) jarvis@jia:~/coding/rust_home/github.com/meua/dora-drives$ pip list | grep dora
dora-rs                              0.2.3rc6
(dora3.7) jarvis@jia:~/coding/rust_home/github.com/meua/dora-drives$ dora -V
dora-cli 0.2.3-rc6
(dora3.7) jarvis@jia:~/coding/rust_home/github.com/meua/dora-drives$ dora start graphs/tutorials/webcam_midas_frame.yaml --attach --hot-reload --name dpt_midas
16f3e1ec-96f4-48e7-9140-85e0e7224fe3

I have tested this function as you described and it is ready. The output of running the task is as follows:

![image](https://user-images.githubusercontent.com/11570442/236610613-e99b291f-bc80-496c-bdff-b12b367c9a2c.png)