Upgrade required version of vineyard. (#2588)

Author: sighingnow

.github/workflows/platform-ci.yml

@@ -71,7 +71,7 @@ jobs:
               conda install -n test --quiet --yes -c conda-forge python=$PYTHON skein conda-pack
             fi
             if [ -n "$WITH_VINEYARD" ]; then
-              pip install vineyard==0.2.7 -i https://pypi.org/simple
+              pip install vineyard -i https://pypi.org/simple
 
               mkdir -p /tmp/etcd-download-test
               export ETCD_VER=v3.4.13

mars/dataframe/datasource/from_vineyard.py

@@ -18,7 +18,7 @@
 from ... import opcodes as OperandDef
 from ...core import OutputType
 from ...core.context import get_context
-from ...serialization.serializables import StringField
+from ...serialization.serializables import Int32Field, StringField
 from ...tensor.datasource.from_vineyard import resolve_vineyard_socket
 from ...utils import calc_nsplits, has_unknown_shape
 from ..operands import DataFrameOperand, DataFrameOperandMixin
@@ -44,6 +44,9 @@ class DataFrameFromVineyard(DataFrameOperand, DataFrameOperandMixin):
     # ObjectID in vineyard
     object_id = StringField("object_id")
 
+    # a dummy attr to make sure ops have different keys
+    operator_index = Int32Field("operator_index")
+
     def __init__(self, vineyard_socket=None, object_id=None, **kw):
         super().__init__(
             vineyard_socket=vineyard_socket,
@@ -92,6 +95,7 @@ def tile(cls, op):
         for index, worker in enumerate(workers):
             chunk_op = op.copy().reset_key()
             chunk_op.expect_worker = worker
+            chunk_op.operator_index = index
             out_chunk = chunk_op.new_chunk(
                 [],
                 dtypes=dtypes,

mars/dataframe/datastore/tests/test_datastore_execution.py

@@ -237,7 +237,7 @@ def test_vineyard_execution(setup):
         "check_index_value": False,
     }
 
-    with vineyard.deploy.local.start_vineyardd() as (_, vineyard_socket):
+    with vineyard.deploy.local.start_vineyardd() as (_, vineyard_socket, _):
         raw = pd.DataFrame({"a": np.arange(0, 55), "b": np.arange(55, 110)})
         a = md.DataFrame(raw, chunk_size=15)
         a.execute()  # n.b.: pre-execute

mars/dataframe/datastore/to_vineyard.py

@@ -18,7 +18,7 @@
 from ... import opcodes as OperandDef
 from ...core import OutputType
 from ...core.operand.base import SchedulingHint
-from ...serialization.serializables import StringField
+from ...serialization.serializables import FieldTypes, StringField, TupleField
 from ...tensor.datastore.to_vineyard import resolve_vineyard_socket
 from ..operands import DataFrameOperand, DataFrameOperandMixin
 from ..utils import parse_index
@@ -37,6 +37,9 @@ class DataFrameToVineyardChunk(DataFrameOperand, DataFrameOperandMixin):
     # vineyard ipc socket
     vineyard_socket = StringField("vineyard_socket")
 
+    # a dummy attr to make sure ops have different keys
+    operator_index = TupleField("operator_index", FieldTypes.int32)
+
     def __init__(self, vineyard_socket=None, dtypes=None, **kw):
         super().__init__(
             vineyard_socket=vineyard_socket,
@@ -75,6 +78,7 @@ def tile(cls, op):
         for idx, chunk in enumerate(op.inputs[0].chunks):
             chunk_op = op.copy().reset_key()
             chunk_op.scheduling_hint = scheduling_hint
+            chunk_op.operator_index = chunk.index
             out_chunk = chunk_op.new_chunk(
                 [chunk],
                 shape=(1, 1),

mars/storage/vineyard.py

@@ -127,6 +127,7 @@ def __init__(self, vineyard_size: int, vineyard_socket: str = None):
     async def setup(cls, **kwargs) -> Tuple[Dict, Dict]:
         loop = asyncio.get_running_loop()
         etcd_endpoints = kwargs.pop("etcd_endpoints", "127.0.0.1:2379")
+        etcd_prefix = kwargs.pop("etcd_prefix", "vineyard")
         vineyard_size = kwargs.pop("vineyard_size", "1Gi")
         vineyard_socket = kwargs.pop("vineyard_socket", None)
         vineyardd_path = kwargs.pop("vineyardd_path", None)
@@ -142,6 +143,7 @@ async def setup(cls, **kwargs) -> Tuple[Dict, Dict]:
         else:
             vineyard_store = vineyard.deploy.local.start_vineyardd(
                 etcd_endpoints,
+                etcd_prefix,
                 vineyardd_path,
                 vineyard_size,
                 vineyard_socket,

mars/tensor/datasource/from_vineyard.py

@@ -16,7 +16,7 @@
 import numpy as np
 
 from ... import opcodes as OperandDef
-from ...serialization.serializables import StringField
+from ...serialization.serializables import Int32Field, StringField
 from ...storage.base import StorageLevel
 from ...utils import calc_nsplits, has_unknown_shape
 from ...core.context import get_context
@@ -50,6 +50,9 @@ class TensorFromVineyard(TensorNoInput):
     # ObjectID in vineyard
     object_id = StringField("object_id")
 
+    # a dummy attr to make sure ops have different keys
+    operator_index = Int32Field("operator_index")
+
     def __init__(self, vineyard_socket=None, object_id=None, **kw):
         super().__init__(vineyard_socket=vineyard_socket, object_id=object_id, **kw)
 
@@ -62,6 +65,7 @@ def tile(cls, op):
         for index, worker in enumerate(workers):
             chunk_op = op.copy().reset_key()
             chunk_op.expect_worker = worker
+            chunk_op.operator_index = index
             out_chunk = chunk_op.new_chunk(
                 [], dtype=np.dtype(object), shape=(1,), index=(index,)
             )
@@ -183,7 +187,7 @@ def fromvineyard(tensor, vineyard_socket=None):
         dtype=np.dtype("byte"),
         gpu=False,
     )
-    meta = metaop(shape=(np.nan,), chunk_size=(1,))
+    meta = metaop(shape=(np.nan,), chunk_size=(np.nan,))
     op = TensorFromVineyardChunk(
         vineyard_socket=vineyard_socket, object_id=object_id, gpu=False
     )

mars/tensor/datastore/tests/test_datastore_execution.py

@@ -223,7 +223,7 @@ def test_vineyard_execution(setup):
         "check_shape": False,
     }
 
-    with vineyard.deploy.local.start_vineyardd() as (_, vineyard_socket):
+    with vineyard.deploy.local.start_vineyardd() as (_, vineyard_socket, _):
         a = tensor(raw, chunk_size=15)
         a.execute()  # n.b.: pre-execute
 

mars/tensor/datastore/to_vineyard.py

@@ -17,7 +17,7 @@
 
 from ... import opcodes as OperandDef
 from ...core.operand.base import SchedulingHint
-from ...serialization.serializables import KeyField, StringField
+from ...serialization.serializables import FieldTypes, KeyField, StringField, TupleField
 from ...storage.base import StorageLevel
 from ..datasource import tensor as astensor
 from .core import TensorDataStore
@@ -52,6 +52,9 @@ class TensorVineyardDataStoreChunk(TensorDataStore):
     # vineyard ipc socket
     vineyard_socket = StringField("vineyard_socket")
 
+    # a dummy attr to make sure ops have different keys
+    operator_index = TupleField("operator_index", FieldTypes.int32)
+
     def __init__(self, vineyard_socket=None, **kw):
         super().__init__(vineyard_socket=vineyard_socket, **kw)
 
@@ -71,6 +74,7 @@ def tile(cls, op):
         for idx, chunk in enumerate(op.inputs[0].chunks):
             chunk_op = op.copy().reset_key()
             chunk_op.scheduling_hint = scheduling_hint
+            chunk_op.operator_index = chunk.index
             out_chunk = chunk_op.new_chunk(
                 [chunk], dtype=np.dtype("O"), shape=(1,), index=(idx,)
             )

setup.cfg

@@ -74,7 +74,7 @@ kubernetes =
 ray =
     ray<1.4
 vineyard =
-    vineyard==0.2.7; sys.platform != "win32"
+    vineyard>=0.3; sys.platform != "win32"
 
 [tool:pytest]
 markers =