Tune CLI modifications

.travis/install-dependencies.sh

@@ -24,7 +24,7 @@ if [[ "$PYTHON" == "2.7" ]] && [[ "$platform" == "linux" ]]; then
   wget https://repo.continuum.io/miniconda/Miniconda2-4.5.4-Linux-x86_64.sh -O miniconda.sh -nv
   bash miniconda.sh -b -p $HOME/miniconda
   export PATH="$HOME/miniconda/bin:$PATH"
-  pip install -q cython==0.29.0 cmake tensorflow gym opencv-python pyyaml pandas==0.23.4 requests \
+  pip install -q cython==0.29.0 cmake tensorflow gym==0.10.11 opencv-python pyyaml pandas==0.23.4 requests \
     feather-format lxml openpyxl xlrd py-spy setproctitle faulthandler pytest-timeout mock flaky
 elif [[ "$PYTHON" == "3.5" ]] && [[ "$platform" == "linux" ]]; then
   sudo apt-get update
@@ -50,7 +50,7 @@ elif [[ "$PYTHON" == "2.7" ]] && [[ "$platform" == "macosx" ]]; then
   wget https://repo.continuum.io/miniconda/Miniconda2-4.5.4-MacOSX-x86_64.sh -O miniconda.sh -nv
   bash miniconda.sh -b -p $HOME/miniconda
   export PATH="$HOME/miniconda/bin:$PATH"
-  pip install -q cython==0.29.0 cmake tensorflow gym opencv-python pyyaml pandas==0.23.4 requests \
+  pip install -q cython==0.29.0 cmake tensorflow gym==0.10.11 opencv-python pyyaml pandas==0.23.4 requests \
     feather-format lxml openpyxl xlrd py-spy setproctitle faulthandler pytest-timeout mock flaky
 elif [[ "$PYTHON" == "3.5" ]] && [[ "$platform" == "macosx" ]]; then
   # check that brew is installed

cmake/Modules/ArrowExternalProject.cmake

@@ -15,17 +15,17 @@
 #  - PLASMA_SHARED_LIB
 
 set(arrow_URL https://github.com/ray-project/arrow.git)
-# This commit is based on https://github.com/apache/arrow/pull/3410. We
+# This commit is based on https://github.com/apache/arrow/pull/3526. We
 # include the link here to make it easier to find the right commit because
 # Arrow often rewrites git history and invalidates certain commits.
 # It has been patched to fix an upstream symbol clash with TensorFlow,
 # the patch is available at
-# https://github.com/ray-project/arrow/commit/511dae1149e3656bbf84f461729f2306d2ebf2e5
+# https://github.com/ray-project/arrow/commit/007e1ca289e979bac80231fa9ee7510be744b60b
 # See the discussion in https://github.com/apache/arrow/pull/3177
 # WARNING: If the arrow version is updated, you need to also update the
 # SETUPTOOLS_SCM_PRETEND_VERSION version string in the ThirdpartyToolchain.cmake
 # file
-set(arrow_TAG 511dae1149e3656bbf84f461729f2306d2ebf2e5)
+set(arrow_TAG 007e1ca289e979bac80231fa9ee7510be744b60b)
 
 set(ARROW_INSTALL_PREFIX ${CMAKE_CURRENT_BINARY_DIR}/external/arrow-install)
 set(ARROW_HOME ${ARROW_INSTALL_PREFIX})

doc/source/example-parameter-server.rst

@@ -123,5 +123,5 @@ resulting gradients.
       current_weights = ps.apply_gradients.remote(*gradients)
 
 Both of these examples implement the parameter server using a single actor,
-however they can be easily extended to **shard the parameters across multiple
+however they can be easily extended to **split the parameters across multiple
 actors**.

docker/examples/Dockerfile

@@ -5,7 +5,7 @@ FROM ray-project/deploy
 # This updates numpy to 1.14 and mutes errors from other libraries
 RUN conda install -y numpy
 RUN apt-get install -y zlib1g-dev
-RUN pip install gym[atari] opencv-python==3.2.0.8 tensorflow lz4 keras pytest-timeout smart_open
+RUN pip install gym[atari]==0.10.11 opencv-python==3.2.0.8 tensorflow lz4 keras pytest-timeout smart_open
 RUN pip install -U h5py  # Mutes FutureWarnings
 RUN pip install --upgrade bayesian-optimization
 RUN pip install --upgrade git+git://github.com/hyperopt/hyperopt.git

java/pom.xml

@@ -29,7 +29,7 @@
       <dependency>
         <groupId>org.apache.arrow</groupId>
         <artifactId>arrow-plasma</artifactId>
-        <version>0.10.0</version>
+        <version>0.13.0-SNAPSHOT</version>
       </dependency>
       <dependency>
         <groupId>de.ruedigermoeller</groupId>

java/runtime/src/main/java/org/ray/runtime/config/RayConfig.java

@@ -6,8 +6,10 @@
 import com.typesafe.config.Config;
 import com.typesafe.config.ConfigException;
 import com.typesafe.config.ConfigFactory;
+import com.typesafe.config.ConfigValue;
 
 import java.io.File;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import org.ray.api.id.UniqueId;
@@ -50,6 +52,7 @@ public class RayConfig {
   public final Long objectStoreSize;
 
   public final String rayletSocketName;
+  public final List<String> rayletConfigParameters;
 
   public final String redisServerExecutablePath;
   public final String redisModulePath;
@@ -162,6 +165,14 @@ public RayConfig(Config config) {
     // raylet socket name
     rayletSocketName = config.getString("ray.raylet.socket-name");
 
+    // raylet parameters
+    rayletConfigParameters = new ArrayList<String>();
+    Config rayletConfig = config.getConfig("ray.raylet.config");
+    for (java.util.Map.Entry<java.lang.String,ConfigValue> entry : rayletConfig.entrySet()) {
+      String parameter = entry.getKey() + "," + String.valueOf(entry.getValue().unwrapped());
+      rayletConfigParameters.add(parameter);
+    }
+
     // library path
     this.libraryPath = new ImmutableList.Builder<String>().add(
         rayHome + "/build/src/plasma",

java/runtime/src/main/java/org/ray/runtime/objectstore/MockObjectStore.java

@@ -1,11 +1,13 @@
 package org.ray.runtime.objectstore;
 
+import java.lang.reflect.Constructor;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import org.apache.arrow.plasma.ObjectStoreLink;
+import org.apache.arrow.plasma.ObjectStoreLink.ObjectStoreData;
 import org.ray.api.id.UniqueId;
 import org.ray.runtime.RayDevRuntime;
 import org.ray.runtime.raylet.MockRayletClient;
@@ -57,12 +59,18 @@ public List<byte[]> get(byte[][] objectIds, int timeoutMs, boolean isMetadata) {
   }
 
   @Override
-  public List<byte[]> wait(byte[][] objectIds, int timeoutMs, int numReturns) {
-    ArrayList<byte[]> rets = new ArrayList<>();
+  public List<ObjectStoreData> get(byte[][] objectIds, int timeoutMs) {
+    ArrayList<ObjectStoreData> rets = new ArrayList<>();
+    // TODO(yuhguo): make ObjectStoreData's constructor public.
     for (byte[] objId : objectIds) {
-      //tod test
-      if (data.containsKey(new UniqueId(objId))) {
-        rets.add(objId);
+      UniqueId uniqueId = new UniqueId(objId);
+      try {
+        Constructor<ObjectStoreData> constructor = ObjectStoreData.class.getConstructor(
+            byte[].class, byte[].class);
+        constructor.setAccessible(true);
+        rets.add(constructor.newInstance(metadata.get(uniqueId), data.get(uniqueId)));
+      } catch (Exception e) {
+        throw new RuntimeException(e);
       }
     }
     return rets;
@@ -73,11 +81,6 @@ public byte[] hash(byte[] objectId) {
     return null;
   }
 
-  @Override
-  public void fetch(byte[][] objectIds) {
-
-  }
-
   @Override
   public long evict(long numBytes) {
     return 0;
@@ -89,8 +92,12 @@ public void release(byte[] objectId) {
   }
 
   @Override
-  public boolean contains(byte[] objectId) {
+  public void delete(byte[] objectId) {
+    return;
+  }
 
+  @Override
+  public boolean contains(byte[] objectId) {
     return data.containsKey(new UniqueId(objectId));
   }
 

java/runtime/src/main/java/org/ray/runtime/objectstore/ObjectStoreProxy.java

@@ -4,6 +4,7 @@
 import java.util.List;
 import org.apache.arrow.plasma.ObjectStoreLink;
 import org.apache.arrow.plasma.PlasmaClient;
+import org.apache.arrow.plasma.exceptions.DuplicateObjectException;
 import org.apache.commons.lang3.tuple.Pair;
 import org.ray.api.exception.RayException;
 import org.ray.api.id.UniqueId;
@@ -12,13 +13,17 @@
 import org.ray.runtime.config.RunMode;
 import org.ray.runtime.util.Serializer;
 import org.ray.runtime.util.UniqueIdUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Object store proxy, which handles serialization and deserialization, and utilize a {@code
  * org.ray.spi.ObjectStoreLink} to actually store data.
  */
 public class ObjectStoreProxy {
 
+  private static final Logger LOGGER = LoggerFactory.getLogger(ObjectStoreProxy.class);
+
   private static final int GET_TIMEOUT_MS = 1000;
 
   private final AbstractRayRuntime runtime;
@@ -82,11 +87,19 @@ public <T> List<Pair<T, GetStatus>> get(List<UniqueId> ids, int timeoutMs, boole
   }
 
   public void put(UniqueId id, Object obj, Object metadata) {
-    objectStore.get().put(id.getBytes(), Serializer.encode(obj), Serializer.encode(metadata));
+    try {
+      objectStore.get().put(id.getBytes(), Serializer.encode(obj), Serializer.encode(metadata));
+    } catch (DuplicateObjectException e) {
+      LOGGER.warn(e.getMessage());
+    }
   }
 
   public void putSerialized(UniqueId id, byte[] obj, byte[] metadata) {
-    objectStore.get().put(id.getBytes(), obj, metadata);
+    try {
+      objectStore.get().put(id.getBytes(), obj, metadata);
+    } catch (DuplicateObjectException e) {
+      LOGGER.warn(e.getMessage());
+    }
   }
 
   public enum GetStatus {

java/runtime/src/main/java/org/ray/runtime/runner/RunManager.java

@@ -205,7 +205,7 @@ private void startRaylet() {
         "0", // number of initial workers
         String.valueOf(maximumStartupConcurrency),
         ResourceUtil.getResourcesStringFromMap(rayConfig.resources),
-        "",  // The internal config list.
+        String.join(",", rayConfig.rayletConfigParameters),  // The internal config list.
         buildPythonWorkerCommand(), // python worker command
         buildWorkerCommandRaylet() // java worker command
     );

java/runtime/src/main/resources/ray.default.conf

@@ -85,6 +85,10 @@ ray {
   raylet {
     // RPC socket name of Raylet
     socket-name: /tmp/ray/sockets/raylet
+
+    // See src/ray/ray_config_def.h for options.
+    config {
+    }
   }
 
 }

java/test/src/main/java/org/ray/api/test/BaseTest.java

@@ -13,6 +13,7 @@ public class BaseTest {
   public void setUp() {
     System.setProperty("ray.home", "../..");
     System.setProperty("ray.resources", "CPU:4,RES-A:4");
+    System.setProperty("ray.raylet.config.inline_object_max_size_bytes", "0");
     Ray.init();
   }
 
@@ -29,6 +30,7 @@ public void tearDown() {
     // unset system properties
     System.clearProperty("ray.home");
     System.clearProperty("ray.resources");
+    System.clearProperty("ray.raylet.config.inline_object_max_size_bytes");
   }
 
 }

python/ray/__init__.py

@@ -51,7 +51,7 @@
 
 from ray._raylet import (UniqueID, ObjectID, DriverID, ClientID, ActorID,
                          ActorHandleID, FunctionID, ActorClassID, TaskID,
-                         Config as _Config)  # noqa: E402
+                         _ID_TYPES, Config as _Config)  # noqa: E402
 
 _config = _Config()
 
@@ -70,14 +70,15 @@
 from ray.actor import method  # noqa: E402
 
 # Ray version string.
-__version__ = "0.6.2"
+__version__ = "0.6.3"
 
 __all__ = [
     "error_info", "init", "connect", "disconnect", "get", "put", "wait",
     "remote", "profile", "actor", "method", "get_gpu_ids", "get_resource_ids",
     "get_webui_url", "register_custom_serializer", "shutdown",
     "is_initialized", "SCRIPT_MODE", "WORKER_MODE", "LOCAL_MODE",
-    "PYTHON_MODE", "global_state", "_config", "__version__", "internal"
+    "PYTHON_MODE", "global_state", "_config", "__version__", "internal",
+    "_ID_TYPES"
 ]
 
 __all__ += [