Object hashes (#104)

* factoring out object_info for general use by several Ray components

* addressing comments

* Replace SHA256 task hash with MD5

Add object hash to object table (always overwrites)

Support for table operations that span multiple asynchronous Redis
commands

Add a new object location in a transaction, using Redis's optimistic
concurrency

Use Redis GETSET instead of transactions and Python frontend code for object hashing

Remove spurious log message

Fix for object_table_add

Revert "Replace SHA256 task hash with MD5"

This reverts commit e599de473c8dad9189ccb0600429534b469b76a2.

Revert to sha256

Test case for illegal puts

Use SETNX to set object hashes

Initialize digest with zeros

Initialize plasma_request with zeros

* Fixes

* replace SHA256 with a faster hash in the object store

* Fix valgrind

* Address Robert's comments

* Check that plasma_compute_object_hash succeeds.

* Don't run test_illegal_put test with valgrind because it causes an intentional crash which causes valgrind to complain.

* Debugging after rebase.

* handling Robert's comments

* Fix bugs after rebase.

* final fixes for Stephanie's PR

* fix

src/plasma/lib/python/plasma.py

@@ -1,6 +1,5 @@
 import os
 import random
-import socket
 import subprocess
 import time
 import libplasma
@@ -135,6 +134,18 @@ class PlasmaClient(object):
     """
     return libplasma.contains(self.conn, object_id)
 
+  def hash(self, object_id):
+    """Compute the hash of an object in the object store.
+
+    Args:
+      object_id (str): A string used to identify an object.
+
+    Returns:
+      A digest string object's SHA256 hash. If the object isn't in the object
+      store, the string will have length zero.
+    """
+    return libplasma.hash(self.conn, object_id)
+
   def seal(self, object_id):
     """Seal the buffer in the PlasmaStore for a particular object ID.
 
@@ -209,27 +220,11 @@ class PlasmaClient(object):
 
   def subscribe(self):
     """Subscribe to notifications about sealed objects."""
-    fd = libplasma.subscribe(self.conn)
-    self.notification_sock = socket.fromfd(fd, socket.AF_UNIX, socket.SOCK_STREAM)
-    self.notification_fd = fd
-    # Make the socket non-blocking.
-    self.notification_sock.setblocking(0)
+    self.notification_fd = libplasma.subscribe(self.conn)
 
   def get_next_notification(self):
     """Get the next notification from the notification socket."""
-    if not self.notification_sock:
-      raise Exception("To get notifications, first call subscribe.")
-    # Loop until we've read PLASMA_ID_SIZE bytes from the socket.
-    while True:
-      try:
-        rv = libplasma.receive_notification(self.notification_fd)
-        obj_id, data_size, metadata_size = rv
-      except socket.error:
-        time.sleep(0.001)
-      else:
-        assert len(obj_id) == PLASMA_ID_SIZE
-        break
-    return obj_id, data_size, metadata_size
+    return libplasma.receive_notification(self.notification_fd)
 
 DEFAULT_PLASMA_STORE_MEMORY = 10 ** 9